File size: 3,688 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | #!/usr/bin/env python3
"""
Import-graph checker — verifies the one-way dependency direction.
Run: python scripts/check_imports.py
Exits 0 if every layer only imports from layers below it.
Exits 1 with a violation report otherwise.
"""
from __future__ import annotations
import importlib
import pkgutil
import sys
from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
# Layer rank — lower number = lower in the stack
LAYER_RANK = {
"models": 0,
"utils": 1,
"config": 1, # config is foundational
"storage": 2,
"metrics": 2, # metrics depends on models only
"providers": 3,
"pipeline": 4,
"orchestrator": 5,
"normalization": 5,
"confidence": 5,
"services": 6,
"api": 7,
}
# Allowed cross-layer imports (lower → higher is forbidden;
# higher → lower is allowed; same-layer is allowed)
# metrics is allowed to import from models only (not storage/providers)
# but we treat it as layer 2 (same as storage) so it can't import storage
def get_layer(module_name: str) -> int | None:
for prefix, rank in LAYER_RANK.items():
if module_name == prefix or module_name.startswith(prefix + "."):
return rank
return None
def check_package(package_name: str) -> list[str]:
"""Import every module in a package and check its imports."""
violations: list[str] = []
try:
pkg = importlib.import_module(package_name)
except ImportError as e:
print(f" SKIP {package_name}: {e}")
return violations
pkg_layer = get_layer(package_name)
if pkg_layer is None:
return violations
# Walk all submodules
if hasattr(pkg, "__path__"):
for _, modname, _ in pkgutil.walk_packages(pkg.__path__, prefix=f"{package_name}."):
try:
mod = importlib.import_module(modname)
except Exception as e:
print(f" SKIP {modname}: {e}")
continue
mod_layer = get_layer(modname)
if mod_layer is None:
continue
# Check every imported module
for imported_name in list(vars(mod).keys()):
# crude: check sys.modules for the actual imported packages
pass
# Better: check mod.__dict__ for module objects
for attr_name, attr_val in vars(mod).items():
if not hasattr(attr_val, "__name__"):
continue
if not isinstance(attr_val, type(sys)): # not a module
continue
imported_module = getattr(attr_val, "__name__", "")
imported_layer = get_layer(imported_module)
if imported_layer is None:
continue
if imported_layer > mod_layer:
violations.append(
f" VIOLATION: {modname} (layer {mod_layer}) imports "
f"{imported_module} (layer {imported_layer})"
)
return violations
def main() -> int:
print("Checking import graph for one-way dependency violations...")
print()
all_violations: list[str] = []
for pkg in LAYER_RANK:
print(f"Checking {pkg}/...")
violations = check_package(pkg)
all_violations.extend(violations)
print()
if all_violations:
print(f"FOUND {len(all_violations)} violation(s):")
for v in all_violations:
print(v)
return 1
else:
print("OK — no dependency-direction violations found.")
return 0
if __name__ == "__main__":
sys.exit(main())
|