Coupling

coupling

Coupling helpers — graph algorithms and config parsing.

Public-internal API: the eight non-underscore functions below are stable enough to be imported directly by tests within this package. They are not re-exported in the package root __all__ because they are tools for rule authors, not application code.

build_coupling_result(fan_out, fan_in, threshold, overrides=None, *, orchestrator_bonus=0, imports_map=None, src_path=None, severity_error_multiplier=_COUPLING_DEFAULT_SEVERITY_MULTIPLIER)

Compute coupling summary from fan-out / fan-in metrics.

Classifies each over-threshold module as "warning" or "error" based on severity_error_multiplier: fan-out above effective_threshold * severity_error_multiplier is an error, otherwise a warning.

Returns a dict with keys max_fan_out, max_fan_in, avg_coupling, n_over_threshold, and over_threshold (list of dicts with module, fan_out, role, effective_threshold, severity).

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

def build_coupling_result(  # noqa: PLR0913
    fan_out: dict[str, int],
    fan_in: dict[str, int],
    threshold: int,
    overrides: dict[str, int] | None = None,
    *,
    orchestrator_bonus: int = 0,
    imports_map: dict[str, list[str]] | None = None,
    src_path: Path | None = None,
    severity_error_multiplier: int = _COUPLING_DEFAULT_SEVERITY_MULTIPLIER,
) -> dict[str, Any]:
    """Compute coupling summary from fan-out / fan-in metrics.

    Classifies each over-threshold module as ``"warning"`` or ``"error"``
    based on *severity_error_multiplier*: fan-out above
    ``effective_threshold * severity_error_multiplier`` is an error,
    otherwise a warning.

    Returns a dict with keys ``max_fan_out``, ``max_fan_in``,
    ``avg_coupling``, ``n_over_threshold``, and ``over_threshold``
    (list of dicts with ``module``, ``fan_out``, ``role``,
    ``effective_threshold``, ``severity``).
    """
    _overrides = overrides or {}
    _imports_map = imports_map or {}

    def _effective_threshold(name: str) -> tuple[int, str]:
        """Return ``(effective_threshold, role)`` for *name*."""
        if name in _overrides:
            return _overrides[name], classify_module_role(
                name,
                _imports_map.get(name, []),
                src_path,
            ) if src_path else "leaf"
        for key, val in _overrides.items():
            if name.endswith(f".{key}") or name == key:
                return val, classify_module_role(
                    name,
                    _imports_map.get(name, []),
                    src_path,
                ) if src_path else "leaf"

        role = "leaf"
        if src_path and orchestrator_bonus:
            role = classify_module_role(
                name,
                _imports_map.get(name, []),
                src_path,
            )
        bonus = orchestrator_bonus if role == "orchestrator" else 0
        return threshold + bonus, role

    over: list[dict[str, Any]] = []
    for name, fo in fan_out.items():
        eff, role = _effective_threshold(name)
        if fo > eff:
            severity = "error" if fo > eff * severity_error_multiplier else "warning"
            over.append(
                {
                    "module": name,
                    "fan_out": fo,
                    "role": role,
                    "effective_threshold": eff,
                    "severity": severity,
                }
            )

    over.sort(key=lambda x: x.get("fan_out", 0), reverse=True)

    return {
        "max_fan_out": max(fan_out.values()),
        "max_fan_in": max(fan_in.values()) if fan_in else 0,
        "avg_coupling": sum(fan_out.values()) / len(fan_out),
        "n_over_threshold": len(over),
        "over_threshold": over,
    }

classify_module_role(module_name, imports, src_path)

Classify a module as "orchestrator" or "leaf".

A module is an orchestrator if it imports from >= 3 distinct sibling subpackages within the project namespace. Only intra-project imports are considered (external/stdlib imports are ignored).

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

def classify_module_role(
    module_name: str,
    imports: list[str],
    src_path: Path,
) -> str:
    """Classify a module as ``"orchestrator"`` or ``"leaf"``.

    A module is an orchestrator if it imports from >= 3 distinct sibling
    subpackages within the project namespace.  Only intra-project imports
    are considered (external/stdlib imports are ignored).
    """
    _min_subpackage_depth = 3
    parts = module_name.split(".")
    if len(parts) < _min_subpackage_depth:
        return "leaf"

    internal_prefixes = _detect_internal_prefixes(src_path)
    siblings = _count_siblings(module_name, imports, internal_prefixes)

    _min_siblings_for_orchestrator = 3
    return "orchestrator" if len(siblings) >= _min_siblings_for_orchestrator else "leaf"

extract_imports(tree)

Extract module-level imported module names from an AST.

Only scans top-level imports to avoid false positives from lazy/deferred imports inside functions (which don't cause circular import issues at runtime).

Counts source modules, not individual imported symbols. For example, from foo import A, B counts as a single import of foo.

__future__ imports are excluded — they are language directives, not real dependencies.

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

def extract_imports(tree: ast.Module) -> list[str]:
    """Extract module-level imported module names from an AST.

    Only scans top-level imports to avoid false positives from lazy/deferred
    imports inside functions (which don't cause circular import issues at runtime).

    Counts source modules, not individual imported symbols. For example,
    ``from foo import A, B`` counts as a single import of ``foo``.

    ``__future__`` imports are excluded — they are language directives,
    not real dependencies.
    """
    imports: list[str] = []
    for node in tree.body:
        if isinstance(node, ast.Import):
            for alias in node.names:
                imports.append(alias.name)
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.module != "__future__":
                imports.append(node.module)
    return imports

parse_overrides(raw)

Parse an overrides mapping, silently dropping invalid entries.

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

def parse_overrides(raw: object) -> dict[str, int]:
    """Parse an overrides mapping, silently dropping invalid entries."""
    if not isinstance(raw, dict):
        return {}
    try:
        return {str(k): int(v) for k, v in raw.items()}
    except (TypeError, ValueError):
        return {}

read_coupling_config(project_path)

Read coupling thresholds from [tool.axm-audit.coupling] in pyproject.toml.

Returns:

Type	Description
int	``(fan_out_threshold, overrides, orchestrator_bonus,
dict[str, int]	severity_error_multiplier)`` — falls back to defaults on any error.

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

def read_coupling_config(
    project_path: Path,
) -> tuple[int, dict[str, int], int, int]:
    """Read coupling thresholds from ``[tool.axm-audit.coupling]`` in pyproject.toml.

    Returns:
        ``(fan_out_threshold, overrides, orchestrator_bonus,
        severity_error_multiplier)`` — falls back to defaults on any error.
    """
    defaults: tuple[int, dict[str, int], int, int] = (
        _COUPLING_DEFAULT_THRESHOLD,
        {},
        _COUPLING_DEFAULT_ORCHESTRATOR_BONUS,
        _COUPLING_DEFAULT_SEVERITY_MULTIPLIER,
    )
    pyproject = project_path / "pyproject.toml"
    if not pyproject.exists():
        return defaults

    try:
        data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
    except Exception:  # noqa: BLE001
        return defaults

    section = data.get("tool", {}).get("axm-audit", {}).get("coupling", {})

    threshold = safe_int(
        section.get("fan_out_threshold", _COUPLING_DEFAULT_THRESHOLD),
        _COUPLING_DEFAULT_THRESHOLD,
    )
    overrides = parse_overrides(section.get("overrides", {}))
    bonus = safe_int(
        section.get("orchestrator_bonus", _COUPLING_DEFAULT_ORCHESTRATOR_BONUS),
        _COUPLING_DEFAULT_ORCHESTRATOR_BONUS,
    )
    multiplier = max(
        safe_int(
            section.get(
                "severity_error_multiplier", _COUPLING_DEFAULT_SEVERITY_MULTIPLIER
            ),
            _COUPLING_DEFAULT_SEVERITY_MULTIPLIER,
        ),
        1,
    )

    return threshold, overrides, bonus, multiplier

safe_int(value, default)

Convert value to a non-negative int, returning default on failure.

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

def safe_int(value: Any, default: int) -> int:
    """Convert *value* to a non-negative ``int``, returning *default* on failure."""
    try:
        result = int(value)
    except (TypeError, ValueError):
        return default
    return result if result >= 0 else default

strip_prefix(modules)

Strip the common top-level package prefix from module names.

All modules in a detected cycle belong to the same package (the graph only contains modules under src/). Removing the shared prefix cuts token count by ~49% with zero information loss.

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

def strip_prefix(modules: list[str]) -> list[str]:
    """Strip the common top-level package prefix from module names.

    All modules in a detected cycle belong to the same package (the graph
    only contains modules under ``src/``).  Removing the shared prefix
    cuts token count by ~49% with zero information loss.
    """
    if not modules:
        return modules
    first_dot = modules[0].find(".")
    if first_dot == -1:
        return modules
    prefix = modules[0][: first_dot + 1]
    if all(m.startswith(prefix) for m in modules):
        return [m[len(prefix) :] for m in modules]
    return modules

tarjan_scc(graph)

Find strongly connected components using iterative Tarjan's algorithm.

Uses an explicit call stack instead of recursion to avoid hitting Python's recursion limit on large graphs (>1000 modules).

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

def tarjan_scc(graph: dict[str, set[str]]) -> list[list[str]]:
    """Find strongly connected components using iterative Tarjan's algorithm.

    Uses an explicit call stack instead of recursion to avoid hitting
    Python's recursion limit on large graphs (>1000 modules).
    """
    state = _TarjanState()

    for root in graph:
        if root in state.index:
            continue

        call_stack: list[tuple[str, Iterator[str]]] = []
        state.enter(root)
        call_stack.append((root, iter(graph.get(root, set()))))

        while call_stack:
            node, neighbors = call_stack[-1]

            if _try_advance(state, node, neighbors, graph, call_stack):
                continue

            if state.lowlink[node] == state.index[node]:
                state.pop_scc(node)

            call_stack.pop()
            if call_stack:
                parent = call_stack[-1][0]
                state.lowlink[parent] = min(state.lowlink[parent], state.lowlink[node])

    return state.sccs