orchestrata_ultimate.verification.autofix.import_fixes

Reference › orchestrata_ultimate › verification › autofix › import_fixes

Import-related fixers for the autofix engine (Spec 031). autofix_relative_imports detects the project package name from a src/ or flat layout (_detect_package_name, _detect_flat_package) and rewrites from .x import and from . import statements into absolute imports, skipping .venv and __pycache__; it returns the number of files changed, or 0 when no package name is found. autofix_missing_stdlib_imports and autofix_missing_barename_imports add imports that a file uses but lacks (for example json, dataclasses, timedelta, Optional); _extend_or_add_import extends an existing from-import when one exists. autofix_import_order runs isort --profile black through the given interpreter with a 30-second timeout, logging and skipping on timeout or a missing interpreter. Files are found with walk_python_files and edited in place.

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/import_fixes.py

"""Import-related autofix functions.

Fixes relative imports, import ordering, missing stdlib imports,
and missing bare-name imports (timedelta, Optional, etc.).

Spec: 007 - Verification Agent
"""

import logging
import os
import re
import subprocess

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

_logger = logging.getLogger(__name__)

_STDLIB_PATTERNS = {
    "dataclasses": (
        r"\bdataclasses\.(FrozenInstanceError|fields|" r"asdict|astuple)\b",
        "import dataclasses",
    ),
    "json": (
        r"\bjson\.(dumps|loads|dump|load)\b",
        "import json",
    ),
}

_BARE_IMPORTS = [
    ("timedelta", r"\btimedelta\s*\(", "datetime", "timedelta"),
    ("Optional", r"\bOptional\s*\[", "typing", "Optional"),
    ("List", r"\bList\s*\[", "typing", "List"),
    ("Dict", r"\bDict\s*\[", "typing", "Dict"),
    ("Any", r"\bAny\b", "typing", "Any"),
    ("Union", r"\bUnion\s*\[", "typing", "Union"),
    ("Tuple", r"\bTuple\s*\[", "typing", "Tuple"),
]


@traces(spec="031", story="US-031-001", reqs=["REQ-031-011"])
def autofix_relative_imports(workspace: str) -> int:
    """Convert relative imports to absolute using src/ layout.

    Args:
        workspace: Root directory of the project.

    Returns:
        Number of files fixed.
    """
    pkg_name = _detect_package_name(workspace)
    if not pkg_name:
        return 0
    fixed = 0
    for fpath in walk_python_files(workspace):
        if ".venv" in fpath or "__pycache__" in fpath:
            continue
        content = _safe_read(fpath)
        if content is None:
            continue
        original = content
        content = re.sub(
            r"from \.([\w.]+) import",
            f"from {pkg_name}.\\1 import",
            content,
        )
        content = re.sub(
            r"from \. import",
            f"from {pkg_name} import",
            content,
        )
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-011"])
def _detect_package_name(workspace: str) -> str:
    """Detect package name from src/ or flat layout."""
    src_dir = os.path.join(workspace, "src")
    if os.path.isdir(src_dir):
        for entry in os.listdir(src_dir):
            entry_path = os.path.join(src_dir, entry)
            if not os.path.isdir(entry_path):
                continue
            if entry.startswith((".", "__")):
                continue
            init = os.path.join(entry_path, "__init__.py")
            if os.path.isfile(init):
                return entry
    return _detect_flat_package(workspace)


@traces(spec="031", story="US-031-001", reqs=["REQ-031-011"])
def _detect_flat_package(workspace: str) -> str:
    """Detect package name from flat layout.

    A workspace that is not a directory has no package to detect: report
    "" (which makes every autofix a no-op) instead of raising. Measured
    2026-10-08: the generated autofix tests pass a non-existent workspace and
    only passed when a stray ``test_value/`` directory happened to exist in
    the runner's cwd — green in isolation, ``FileNotFoundError`` in a batch.
    """
    if not os.path.isdir(workspace):
        return ""
    skip = (".", "__", "tests", "design", "docs")
    for entry in os.listdir(workspace):
        entry_path = os.path.join(workspace, entry)
        if not os.path.isdir(entry_path):
            continue
        if entry.startswith(skip):
            continue
        init = os.path.join(entry_path, "__init__.py")
        if os.path.isfile(init):
            return entry
    return ""


@traces(spec="031", story="US-031-001", reqs=["REQ-031-013"])
def autofix_import_order(
    workspace: str,
    venv_python: str = "python3",
) -> None:
    """Auto-fix import ordering with isort.

    Args:
        workspace: Root directory of the project.
        venv_python: Path to the Python interpreter.
    """
    try:
        subprocess.run(
            [venv_python, "-m", "isort", "--profile", "black", "."],
            cwd=workspace,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
        _logger.debug("isort autofix skipped: %s", exc)


@traces(spec="031", story="US-031-001", reqs=["REQ-031-005"])
def autofix_missing_stdlib_imports(workspace: str) -> int:
    """Add missing stdlib imports (json, dataclasses).

    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
        original = content
        content = _add_stdlib_imports(content, original)
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-005"])
def _add_stdlib_imports(
    content: str,
    original: str,
) -> str:
    """Add missing stdlib import statements."""
    for module, (usage_re, import_stmt) in _STDLIB_PATTERNS.items():
        if not re.search(usage_re, content):
            continue
        if re.search(
            rf"^import {module}\b|^from {module} import",
            content,
            re.MULTILINE,
        ):
            continue
        before = content
        content = re.sub(
            r"^(import \w+.*\n)",
            rf"\1{import_stmt}\n",
            content,
            count=1,
            flags=re.MULTILINE,
        )
        if content == before:
            content = import_stmt + "\n" + content
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-006"])
def autofix_missing_barename_imports(workspace: str) -> int:
    """Add missing bare-name imports (timedelta, Optional, etc.).

    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
        original = content
        content = _add_barename_imports(content, original)
        if content != original:
            _safe_write(fpath, content)
            fixed += 1
    return fixed


@traces(spec="031", story="US-031-001", reqs=["REQ-031-006"])
def _add_barename_imports(
    content: str,
    original: str,
) -> str:
    """Add missing bare-name from-import statements."""
    for name, usage_re, module, import_name in _BARE_IMPORTS:
        if not re.search(usage_re, content):
            continue
        if re.search(
            rf"^from {module} import .*\b{name}\b",
            content,
            re.MULTILINE,
        ):
            continue
        if re.search(
            rf"^import {module}$",
            content,
            re.MULTILINE,
        ):
            continue
        content = _extend_or_add_import(
            content,
            original,
            module,
            import_name,
        )
    return content


@traces(spec="031", story="US-031-001", reqs=["REQ-031-006"])
def _extend_or_add_import(
    content: str,
    original: str,
    module: str,
    import_name: str,
) -> str:
    """Extend existing from-import or add a new one."""
    existing = re.search(
        rf"^(from {module} import .+)$",
        content,
        re.MULTILINE,
    )
    if existing:
        old_line = existing.group(1)
        new_line = old_line + f", {import_name}"
        return content.replace(old_line, new_line, 1)
    stmt = f"from {module} import {import_name}"
    result = re.sub(
        r"^(import \w+.*\n)",
        rf"\1{stmt}\n",
        content,
        count=1,
        flags=re.MULTILINE,
    )
    if result == content:
        return stmt + "\n" + content
    return result


@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)