orchestrata_ultimate.verification.autofix.redis_fixes

Reference › orchestrata_ultimate › verification › autofix › redis_fixes

Redis-related fixers for the autofix engine, the largest of the fix modules (Spec 031). Each fixer scans the workspace with walk_python_files, rewrites matching source with regular expressions and returns the number of files changed. The fixers are: autofix_aioredis (the deprecated aioredis style to redis.asyncio), autofix_sync_redis (redis.Redis to redis.asyncio.Redis), autofix_redis_decode (ensure clients use decode_responses=True), autofix_redis_host (replace hardcoded localhost and port in test files and conftest.py, adding import os when needed), autofix_xadd_dict (wrap XADD field dictionaries so values are strings) and autofix_hset_dict (the same for HSET mappings). These target the typical defects of generated code that talks to Redis. Only text is rewritten; no Redis connection is made.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/autofix/redis_fixes.py

"""Redis-related autofix functions.

Fixes deprecated redis imports, sync Redis usage, missing
decode_responses, hardcoded localhost, XADD dict encoding,
and HSET dict encoding.

Spec: 007 - Verification Agent
"""

import os
import re

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.autofix.helpers import (
    walk_python_files,
)


@traces(spec="031", story="US-031-001", reqs=["REQ-031-001", "REQ-031-002"])
def autofix_aioredis(workspace: str) -> int:
    """Replace deprecated redis with redis.asyncio.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None or "redis" not in content:
            continue
        original = content
        content = _replace_aioredis_imports(content)
        content = _replace_aioredis_usage(content)
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-001"])
def _replace_aioredis_imports(content: str) -> str:
    """Replace redis import statements."""
    content = re.sub(
        r"^import redis$",
        "import redis.asyncio as redis",
        content,
        flags=re.MULTILINE,
    )
    content = re.sub(
        r"^from redis import",
        "from redis.asyncio import",
        content,
        flags=re.MULTILINE,
    )
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-002"])
def _replace_aioredis_usage(content: str) -> str:
    """Replace redis usage patterns."""
    for old, new in [
        ("redis.from_url", "redis.from_url"),
        ("redis.Redis", "redis.Redis"),
        ("redis.ConnectionError", "redis.ConnectionError"),
        ("redis.ResponseError", "redis.ResponseError"),
    ]:
        content = content.replace(old, new)
    content = re.sub(r"\baioredis\b", "redis", content)
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-003"])
def autofix_sync_redis(workspace: str) -> int:
    """Replace sync redis.Redis with redis.asyncio.Redis.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None:
            continue
        if "redis.asyncio" in content:
            continue
        if not re.search(
            r"^import redis\b",
            content,
            re.MULTILINE,
        ):
            continue
        if "redis.Redis(" not in content:
            continue
        original = content
        content = _apply_sync_redis_fix(content, original)
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-003"])
def _apply_sync_redis_fix(
    content: str,
    original: str,
) -> str:
    """Apply sync-to-async Redis conversion."""
    content = re.sub(
        r"^import redis$",
        "from redis.asyncio import Redis",
        content,
        flags=re.MULTILINE,
    )
    content = content.replace("redis.Redis(", "Redis(")
    content = re.sub(r"\bredis\.Redis\b", "Redis", content)
    if "redis.exceptions" in content and (
        "from redis" not in content or "exceptions" not in content
    ):
        content = re.sub(
            r"^(from redis\.asyncio import Redis)$",
            r"\1\nfrom redis.exceptions import RedisError",
            content,
            count=1,
            flags=re.MULTILINE,
        )
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-009"])
def autofix_redis_decode(workspace: str) -> int:
    """Ensure Redis clients use decode_responses=True.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None:
            continue
        if "Redis(" not in content:
            continue
        if "decode_responses" in content:
            continue
        original = content
        content = re.sub(
            r"(redis\.(?:async)?Redis\([^)]*)\)",
            r"\1, decode_responses=True)",
            content,
        )
        content = content.replace(
            "(, decode_responses",
            "(decode_responses",
        )
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-010"])
def autofix_redis_host(workspace: str) -> int:
    """Replace hardcoded localhost Redis in test files.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        fname = os.path.basename(fpath)
        if not (fname.startswith("test_") or fname == "conftest.py"):
            continue
        content = _safe_read(fpath)
        if content is None:
            continue
        original = content
        content = _replace_hardcoded_host(content)
        if content != original:
            if "import os" not in content:
                content = "import os\n" + content
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-010"])
def _replace_hardcoded_host(content: str) -> str:
    """Replace hardcoded localhost/port in Redis calls."""
    content = re.sub(
        r"""host\s*=\s*["']localhost["']""",
        "host=os.environ['REDIS_HOST']",
        content,
    )
    content = re.sub(
        r"""Redis\(\s*["']localhost["']""",
        "Redis(os.environ['REDIS_HOST']",
        content,
    )
    content = re.sub(
        r"port\s*=\s*(?:6379|6380)\b",
        "port=int(os.environ['REDIS_PORT'])",
        content,
    )
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-007"])
def autofix_xadd_dict(workspace: str) -> int:
    """Fix XADD calls that pass dicts with non-string values.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None or "xadd" not in content.lower():
            continue
        original = content
        content = re.sub(
            r"\.xadd\(\s*(\w+)\s*,\s*\*\*(\w+)\.to_dict\(\)",
            r".xadd(\1, {k: str(v) for k, v in" r" \2.to_dict().items()}",
            content,
        )
        content = re.sub(
            r"\.xadd\(\s*(\w+)\s*,\s*\*\*(\w+)\s*\)",
            r".xadd(\1, {k: str(v) for k, v in" r" \2.items()})",
            content,
        )
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-008"])
def autofix_hset_dict(workspace: str) -> int:
    """Fix HSET mapping with non-string values.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    fixed = 0
    for fpath in walk_python_files(workspace):
        content = _safe_read(fpath)
        if content is None or "hset" not in content.lower():
            continue
        original = content
        content = re.sub(
            r"\.hset\(([^,]+),\s*mapping\s*=\s*" r"(\w+)\.to_dict\(\)\s*\)",
            r".hset(\1, mapping=" r"{k: str(v) for k, v in \2.to_dict().items()})",
            content,
        )
        content = re.sub(
            r"\.hset\(([^,]+),\s*mapping\s*=\s*\{"
            r"\s*\*\*(\w+)\.to_dict\(\)\s*\}\s*\)",
            r".hset(\1, mapping=" r"{k: str(v) for k, v in \2.to_dict().items()})",
            content,
        )
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
def _safe_read(fpath: str) -> str | None:
    """Read file, returning None on error."""
    try:
        with open(fpath, encoding="utf-8") as f:
            return f.read()
    except OSError, UnicodeDecodeError:
        return None


@traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
def _safe_write(fpath: str, content: str) -> None:
    """Write content to file."""
    with open(fpath, "w", encoding="utf-8") as f:
        f.write(content)