"""Detect first-party module-load import cycles via Tarjan's SCC. Walks every first-party Python module in the repo, builds an import graph from **top-level** ``import`` / ``from ... import`` statements only (function-level lazy imports are intentional runtime breaks and not counted), then reports any strongly-connected component of size > 1, plus any single-module self-loop. Used by ``make check-imports`` (via ``check_imports``) locally and by CI. Exit codes: 0 — zero cycles found 1 — at least one cycle found (output lists every SCC + its edges) How to break a cycle ==================== Two patterns work. Pick by what consumers do with the name. 1. ``import pkg.sub as sub`` (preferred when consumers monkeypatch) Switch :: from pkg.sub import name # binds ``name`` at import time ... name(args) # uses the bound reference to :: import pkg.sub as sub # imports the submodule directly ... sub.name(args) # attribute lookup at call time Both break the static cycle (no edge from the consumer back to the ``pkg`` package). The second form keeps attribute-lookup semantics, which matters whenever consumers monkeypatch ``pkg.sub.name`` in tests — the patched attribute IS looked up at each call. The first form binds the name at import time and ignores later patching. 2. Port / Protocol (when the cycle crosses architectural layers) When the cycle is ``layerA <-> layerB`` (e.g. analytics ↔ sentry, integrations ↔ services), neither side should depend on the other directly. Extract a third module — a ``Protocol`` or a small abstract dataclass — that both depend on, and inject the concrete implementation at startup. See the verifier plugin-registry refactor (``integrations/verification/registry.py``) for the canonical example. Avoid ``from pkg import sub`` (where ``sub`` is a submodule of ``pkg``) inside that ``pkg`` itself or any of its children — that's the form this script flags. It triggers ``pkg``'s ``__init__`` even when you just want the submodule, and re-export patterns in ``__init__`` close the loop. Function-local imports are NOT flagged. They're a legitimate Python pattern for startup-cost deferral (heavy modules in click subcommand bodies), optional dependencies, and conditional / platform-specific code paths. Use sparingly — keep top-level the default — and comment the *why* so future readers don't mistake them for cycle workarounds. """ from __future__ import annotations import ast import sys from collections import defaultdict from collections.abc import Iterable from functools import lru_cache from pathlib import Path # Directories at the repo root that are never first-party import roots. _SKIP_ROOT_DIRS = frozenset( { ".git", ".github", ".pytest_cache", ".ruff_cache", ".venv", "__pycache__", "docs", "opensre.egg-info", "packaging", "tests", "venv", } ) @lru_cache(maxsize=1) def discover_first_party_roots(repo_root: Path | None = None) -> tuple[str, ...]: """Return top-level package names that contain importable Python code.""" root = repo_root or Path(__file__).resolve().parents[2] names: list[str] = [] for child in sorted(root.iterdir()): if not child.is_dir() or child.name.startswith("."): continue if child.name in _SKIP_ROOT_DIRS: continue if not any(child.rglob("*.py")): continue names.append(child.name) return tuple(names) def _top_level_imports(source: str, *, first_party_roots: frozenset[str]) -> set[str]: """Return first-party module paths imported at the module top level. Function-bodies, class-bodies, conditional / try-except wrappers all count as top-level if they are direct module statements — the only imports skipped are those nested **inside a function or class body**. A lazy ``from X import Y`` inside a function does not deadlock at module load, so it should not be flagged as a cycle. """ try: tree = ast.parse(source) except SyntaxError: return set() names: set[str] = set() def _add(module_path: str) -> None: top = module_path.split(".", 1)[0] if top in first_party_roots: names.add(module_path) def _walk_top(body: Iterable[ast.stmt]) -> None: for node in body: if isinstance(node, ast.Import): for alias in node.names: _add(alias.name) elif isinstance(node, ast.ImportFrom): if node.level or not node.module: continue _add(node.module) elif isinstance(node, ast.If): _walk_top(node.body) _walk_top(node.orelse) elif isinstance(node, ast.Try | ast.TryStar): # ``ast.TryStar`` (Python 3.11+, ``try/except*``) shares # the same handler/orelse/finalbody shape as ``ast.Try``. _walk_top(node.body) for handler in node.handlers: _walk_top(handler.body) _walk_top(node.orelse) _walk_top(node.finalbody) elif isinstance(node, ast.With | ast.AsyncWith): _walk_top(node.body) _walk_top(tree.body) return names def _nested_imports(source: str, *, first_party_roots: frozenset[str]) -> list[tuple[str, int]]: """Return ``(target_module, lineno)`` for imports inside function/class bodies. Complements ``_top_level_imports``: catches lazy imports that bypass the module-level direct-edge checker while still being layering violations. """ try: tree = ast.parse(source) except SyntaxError: return [] results: list[tuple[str, int]] = [] def _add(module_path: str, lineno: int) -> None: top = module_path.split(".", 1)[0] if top in first_party_roots: results.append((module_path, lineno)) def _walk_nested(body: Iterable[ast.stmt], *, nested: bool) -> None: for node in body: if isinstance(node, ast.Import): if nested: for alias in node.names: _add(alias.name, node.lineno) elif isinstance(node, ast.ImportFrom): if nested and not node.level and node.module: _add(node.module, node.lineno) elif isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): _walk_nested(node.body, nested=True) elif isinstance(node, ast.If): _walk_nested(node.body, nested=nested) _walk_nested(node.orelse, nested=nested) elif isinstance(node, ast.Try | ast.TryStar): _walk_nested(node.body, nested=nested) for handler in node.handlers: _walk_nested(handler.body, nested=nested) _walk_nested(node.orelse, nested=nested) _walk_nested(node.finalbody, nested=nested) elif isinstance(node, ast.With | ast.AsyncWith): _walk_nested(node.body, nested=nested) elif isinstance(node, ast.For | ast.AsyncFor | ast.While): _walk_nested(node.body, nested=nested) _walk_nested(node.orelse, nested=nested) elif isinstance(node, ast.Match): for case in node.cases: _walk_nested(case.body, nested=nested) _walk_nested(tree.body, nested=False) return results def module_from_path(root: Path, py: Path) -> str: """Resolve a repo-relative ``.py`` path to its dotted module name.""" module = ".".join(py.with_suffix("").relative_to(root).parts) return module.removesuffix(".__init__") def _build_graph(root: Path, first_party_roots: tuple[str, ...]) -> dict[str, set[str]]: """Build the first-party module-level import graph rooted at ``root``.""" roots = frozenset(first_party_roots) graph: dict[str, set[str]] = defaultdict(set) for pkg in first_party_roots: pkg_path = root / pkg if not pkg_path.exists(): continue for py in pkg_path.rglob("*.py"): if "__pycache__" in py.parts: continue module = module_from_path(root, py) source = py.read_text(encoding="utf-8") graph[module].update(_top_level_imports(source, first_party_roots=roots)) return graph def _tarjan_sccs(graph: dict[str, set[str]]) -> list[list[str]]: """Return every strongly-connected component of size > 1, plus any single-module self-loop.""" index: dict[str, int] = {} lowlink: dict[str, int] = {} on_stack: dict[str, bool] = {} stack: list[str] = [] sccs: list[list[str]] = [] counter = [0] def strongconnect(v: str) -> None: index[v] = counter[0] lowlink[v] = counter[0] counter[0] += 1 stack.append(v) on_stack[v] = True for w in graph.get(v, ()): if w not in index: strongconnect(w) lowlink[v] = min(lowlink[v], lowlink[w]) elif on_stack.get(w): lowlink[v] = min(lowlink[v], index[w]) if lowlink[v] == index[v]: component: list[str] = [] while True: w = stack.pop() on_stack[w] = False component.append(w) if w == v: break # ``while True`` above guarantees ``component`` is non-empty here. if len(component) > 1 or component[0] in graph.get(component[0], ()): sccs.append(component) sys.setrecursionlimit(10000) for vertex in list(graph.keys()): if vertex not in index: strongconnect(vertex) return sccs def _format_scc(scc: list[str], graph: dict[str, set[str]]) -> str: """Format an SCC for human-readable output: members + edges within.""" members = sorted(scc) in_scc = set(scc) is_self_loop = len(scc) == 1 edges: list[str] = [] for module in members: for target in sorted(graph.get(module, ())): if target not in in_scc: continue # Single-module self-loops have only the self-edge; show it # explicitly so the developer knows which import closes the # loop. Multi-module SCCs hide self-edges to keep the diff # focused on the cross-module edges that close the cycle. if target == module and not is_self_loop: continue edges.append(f" {module} -> {target}") lines = [f" Modules ({len(scc)}):"] lines.extend(f" - {m}" for m in members) if is_self_loop and not edges: lines.append(" Self-import detected (module imports itself at top level).") elif edges: lines.append(" Edges within SCC:") lines.extend(edges) return "\n".join(lines) def main(argv: list[str] | None = None) -> int: del argv root = Path(__file__).resolve().parents[2] first_party_roots = discover_first_party_roots(root) graph = _build_graph(root, first_party_roots) sccs = _tarjan_sccs(graph) if not sccs: print( f"No import cycles found across {len(graph)} first-party modules " f"({len(first_party_roots)} roots)." ) return 0 print(f"FAIL: {len(sccs)} import cycle(s) found across {len(graph)} first-party modules.") for i, scc in enumerate(sorted(sccs, key=lambda s: -len(s)), 1): print(f"\n## SCC #{i} ({len(scc)} module{'s' if len(scc) > 1 else ''}):") print(_format_scc(scc, graph)) print( "\nTo break a cycle, prefer:\n" " import pkg.sub as sub\n" " ...\n" " sub.name(args)\n" "over:\n" " from pkg.sub import name\n" " ...\n" " name(args)\n" "\n" "Both fix the static cycle; only the first keeps attribute-lookup\n" "semantics so tests that monkeypatch ``pkg.sub.name`` still work.\n" "\n" "For cross-layer cycles, introduce a Protocol/port both sides\n" "depend on (see ``integrations/verification/registry.py`` for the\n" "canonical example).\n" "\n" "Full pattern reference: docstring at the top of this script." ) return 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))