|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Detect circular imports under src/quilt_mcp. |
| 3 | +
|
| 4 | +Prints cycles found among local quilt_mcp modules and exits non-zero when any |
| 5 | +cycle exists. This keeps cycle checks lightweight and CI-friendly. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import ast |
| 11 | +import sys |
| 12 | +from pathlib import Path |
| 13 | +from typing import Dict, List, Set |
| 14 | + |
| 15 | + |
| 16 | +ROOT = Path(__file__).resolve().parents[1] |
| 17 | +SRC_ROOT = ROOT / "src" / "quilt_mcp" |
| 18 | + |
| 19 | + |
| 20 | +def module_name_from_path(path: Path) -> str: |
| 21 | + rel = path.relative_to(ROOT / "src") |
| 22 | + if rel.name == "__init__.py": |
| 23 | + rel = rel.parent |
| 24 | + else: |
| 25 | + rel = rel.with_suffix("") |
| 26 | + return ".".join(rel.parts) |
| 27 | + |
| 28 | + |
| 29 | +def resolve_import(from_module: str, imported: str) -> str: |
| 30 | + if imported.startswith("quilt_mcp."): |
| 31 | + return imported |
| 32 | + if imported == "quilt_mcp": |
| 33 | + return imported |
| 34 | + if imported.startswith("."): |
| 35 | + return "" |
| 36 | + return imported |
| 37 | + |
| 38 | + |
| 39 | +def collect_modules() -> Dict[str, Path]: |
| 40 | + modules: Dict[str, Path] = {} |
| 41 | + for path in SRC_ROOT.rglob("*.py"): |
| 42 | + # Skip package export aggregators. They intentionally re-export symbols |
| 43 | + # and create noisy pseudo-cycles that are not runtime dependencies. |
| 44 | + if path.name == "__init__.py": |
| 45 | + continue |
| 46 | + modules[module_name_from_path(path)] = path |
| 47 | + return modules |
| 48 | + |
| 49 | + |
| 50 | +def extract_imports(module: str, path: Path) -> Set[str]: |
| 51 | + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) |
| 52 | + imports: Set[str] = set() |
| 53 | + |
| 54 | + # Only consider module-level imports. Function-local lazy imports are an |
| 55 | + # intentional pattern used to avoid runtime import cycles. |
| 56 | + for node in tree.body: |
| 57 | + if isinstance(node, ast.Import): |
| 58 | + for alias in node.names: |
| 59 | + target = resolve_import(module, alias.name) |
| 60 | + if target.startswith("quilt_mcp"): |
| 61 | + imports.add(target) |
| 62 | + elif isinstance(node, ast.ImportFrom): |
| 63 | + if node.level > 0: |
| 64 | + pkg_parts = module.split(".") |
| 65 | + base_parts = pkg_parts[:-node.level] |
| 66 | + if node.module: |
| 67 | + base = ".".join(base_parts + node.module.split(".")) |
| 68 | + else: |
| 69 | + base = ".".join(base_parts) |
| 70 | + else: |
| 71 | + base = node.module or "" |
| 72 | + |
| 73 | + target = resolve_import(module, base) |
| 74 | + if target.startswith("quilt_mcp"): |
| 75 | + imports.add(target) |
| 76 | + |
| 77 | + return imports |
| 78 | + |
| 79 | + |
| 80 | +def reduce_to_known_modules(modules: Dict[str, Path], imports: Set[str]) -> Set[str]: |
| 81 | + known = set(modules.keys()) |
| 82 | + reduced: Set[str] = set() |
| 83 | + for imp in imports: |
| 84 | + if imp in known: |
| 85 | + reduced.add(imp) |
| 86 | + continue |
| 87 | + parts = imp.split(".") |
| 88 | + while parts: |
| 89 | + candidate = ".".join(parts) |
| 90 | + if candidate in known: |
| 91 | + reduced.add(candidate) |
| 92 | + break |
| 93 | + parts.pop() |
| 94 | + return reduced |
| 95 | + |
| 96 | + |
| 97 | +def build_graph(modules: Dict[str, Path]) -> Dict[str, Set[str]]: |
| 98 | + graph: Dict[str, Set[str]] = {m: set() for m in modules} |
| 99 | + for mod, path in modules.items(): |
| 100 | + raw_imports = extract_imports(mod, path) |
| 101 | + graph[mod] = reduce_to_known_modules(modules, raw_imports) |
| 102 | + return graph |
| 103 | + |
| 104 | + |
| 105 | +def find_cycles(graph: Dict[str, Set[str]]) -> List[List[str]]: |
| 106 | + cycles: Set[tuple[str, ...]] = set() |
| 107 | + visiting: Set[str] = set() |
| 108 | + visited: Set[str] = set() |
| 109 | + stack: List[str] = [] |
| 110 | + |
| 111 | + def dfs(node: str) -> None: |
| 112 | + visiting.add(node) |
| 113 | + stack.append(node) |
| 114 | + |
| 115 | + for nxt in graph.get(node, set()): |
| 116 | + if nxt in visiting: |
| 117 | + idx = stack.index(nxt) |
| 118 | + cycle = stack[idx:] + [nxt] |
| 119 | + # Canonicalize for deduping. |
| 120 | + core = cycle[:-1] |
| 121 | + min_idx = min(range(len(core)), key=lambda i: core[i]) |
| 122 | + rotated = core[min_idx:] + core[:min_idx] |
| 123 | + cycles.add(tuple(rotated)) |
| 124 | + elif nxt not in visited: |
| 125 | + dfs(nxt) |
| 126 | + |
| 127 | + stack.pop() |
| 128 | + visiting.remove(node) |
| 129 | + visited.add(node) |
| 130 | + |
| 131 | + for module in sorted(graph.keys()): |
| 132 | + if module not in visited: |
| 133 | + dfs(module) |
| 134 | + |
| 135 | + result = [list(c) + [c[0]] for c in sorted(cycles)] |
| 136 | + return result |
| 137 | + |
| 138 | + |
| 139 | +def main() -> int: |
| 140 | + if not SRC_ROOT.exists(): |
| 141 | + print(f"Source root not found: {SRC_ROOT}", file=sys.stderr) |
| 142 | + return 2 |
| 143 | + |
| 144 | + modules = collect_modules() |
| 145 | + graph = build_graph(modules) |
| 146 | + cycles = find_cycles(graph) |
| 147 | + |
| 148 | + if not cycles: |
| 149 | + print("No import cycles detected.") |
| 150 | + return 0 |
| 151 | + |
| 152 | + print(f"Detected {len(cycles)} import cycle(s):") |
| 153 | + for idx, cycle in enumerate(cycles, start=1): |
| 154 | + print(f"{idx:>2}. {' -> '.join(cycle)}") |
| 155 | + return 1 |
| 156 | + |
| 157 | + |
| 158 | +if __name__ == "__main__": |
| 159 | + raise SystemExit(main()) |
0 commit comments