helpers

_helpers

Shared AST helpers for rule implementations.

ASTCache

Thread-safe AST parse cache.

Parses each file at most once per audit session. Uses double-checked locking so multiple threads hitting the same file don't duplicate work.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

class ASTCache:
    """Thread-safe AST parse cache.

    Parses each file at most once per audit session.  Uses
    double-checked locking so multiple threads hitting the same
    file don't duplicate work.
    """

    def __init__(self) -> None:
        self._cache: dict[Path, ast.Module | None] = {}
        self._lock = threading.Lock()

    def get_or_parse(self, path: Path) -> ast.Module | None:
        """Return cached AST or parse *path* and cache the result."""
        resolved = path.resolve()
        if resolved in self._cache:
            return self._cache[resolved]
        with self._lock:
            # Double-check after acquiring lock
            if resolved not in self._cache:
                self._cache[resolved] = parse_file_safe(resolved)
        return self._cache[resolved]

get_or_parse(path)

Return cached AST or parse path and cache the result.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

def get_or_parse(self, path: Path) -> ast.Module | None:
    """Return cached AST or parse *path* and cache the result."""
    resolved = path.resolve()
    if resolved in self._cache:
        return self._cache[resolved]
    with self._lock:
        # Double-check after acquiring lock
        if resolved not in self._cache:
            self._cache[resolved] = parse_file_safe(resolved)
    return self._cache[resolved]

get_ast_cache()

Return the active ASTCache, or None outside audits.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

def get_ast_cache() -> ASTCache | None:
    """Return the active ``ASTCache``, or ``None`` outside audits."""
    return _active_cache

get_python_files(directory)

Get all Python files in a directory recursively.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

def get_python_files(directory: Path) -> list[Path]:
    """Get all Python files in a directory recursively."""
    if not directory.exists():
        return []
    return list(directory.rglob("*.py"))

parse_file_safe(path)

Parse a Python file, returning None on error.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

def parse_file_safe(path: Path) -> ast.Module | None:
    """Parse a Python file, returning None on error."""
    try:
        return ast.parse(path.read_text(), filename=str(path))
    except (SyntaxError, UnicodeDecodeError):
        return None

set_ast_cache(cache)

Set the module-level ASTCache for rule access.

Source code in packages/axm-audit/src/axm_audit/core/rules/_helpers.py

def set_ast_cache(cache: ASTCache | None) -> None:
    """Set the module-level ``ASTCache`` for rule access."""
    global _active_cache
    _active_cache = cache