scripts.architecture_checker
Reference › scripts › architecture_checker
Validates import rules, module boundaries and layering constraints across the main source package. tests/scripts/test_architecture_checker.py imports it; its sibling architecture_static_check takes the same path argument and adds a baseline ratchet. For the layers themselves, see the architecture overview.
Invocation: main takes an optional path (default src/orchestrata_ultimate), -v/--verbose for per-violation detail, -o/--output FILE to write the report, and --strict. Without --strict it exits 0 even when violations exist, so the gate passes; with --strict any violation gives a non-zero exit. A missing path is an error.
check_project parses every module and applies several checks. get_layer maps a module to a layer number and check_import flags an import that breaks the layering; protocol imports and re-exports from an __init__ are exempt (is_protocol_import, is_internal_init_reexport). Imports inside TYPE_CHECKING blocks, and imports scoped to a function, are handled separately: function-scoped cross-layer imports are counted and reported as an accepted pattern. Three further checks look at construction: check_di_instantiation finds code that instantiates the dependency-injection container outside its package, check_direct_driver_construction finds drivers built directly instead of through the factory, and check_graph_seam finds breaches of the graph seam. Test paths are excluded. The report lists files checked, the violation count and each violation.
At a glance
| Source | scripts/architecture_checker.py |
| Lines of code | 691 |
| Classes | 0 |
| Functions | 22 |
| Imports (internal) | none |
| Imported by | none |
| Tests that import it | tests/scripts/test_architecture_checker.py |
Class diagram
No classes; see Functions.
Dependency diagram
No internal imports and no importers.
API
Constants
| Name | Type | Value |
|---|---|---|
LAYER_ASSIGNMENTS |
dict[str, int] |
{'core.domain': 0, 'core.protocols': 0, 'core.packet': 0, 'core.enums': 0, 'cor… |
UPWARD_IMPORT_LIMITS |
dict[int, int] |
{0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5} |
KNOWN_VIOLATIONS |
set[tuple[str, str]] |
{('orchestrata_ultimate.cxp.prolog_bridge', 'orchestrata_ultimate.orchestrator.… |
DI_CONTAINER_HOME_PACKAGES |
frozenset[str] |
frozenset({'plugin', 'gateway', 'di'}) |
_GRAPH_SEAM_ALLOWED |
tuple[str, ...] |
('core/graph_factory.py', 'core/graph/repository.py', 'core/graph/adapters/bolt… |
Functions
| Function | Signature | Summary |
|---|---|---|
_is_test_path |
_is_test_path(path: Path) -> bool |
True if path lives under a tests directory or is a test module. |
_di_container_top_level_package |
_di_container_top_level_package(path: Path) -> str | None |
Return the top-level orchestrata_ultimate.<pkg> package name for a source file path, or None if it isn't under an orchestrata_ultimate tree at all (e.g. a synthetic fixture rooted directly… |
_instantiates_di_container |
_instantiates_di_container(tree: ast.AST) -> bool |
True if the AST contains a call shaped like DIContainer(...) or some.path.DIContainer(...). |
check_di_instantiation |
check_di_instantiation(py_file: Path) -> str | None |
Return a violation description if py_file instantiates DIContainer(...) outside the plugin/gateway/di packages, else None. |
_constructs_driver_directly |
_constructs_driver_directly(tree: ast.AST) -> str | None |
Return a short description of the first direct driver construction in the AST — GraphDatabase.driver(...) / AsyncGraphDatabase.driver(...) or redis.Redis(...) /… |
check_direct_driver_construction |
check_direct_driver_construction(py_file: Path) -> str | None |
Return a violation if py_file constructs a graph/Redis client directly instead of via the core factories (D12), else None. |
_import_probe_lines |
_import_probe_lines(tree: ast.AST) -> set[int] |
Line numbers inside try: bodies that have an ImportError (or ModuleNotFoundError, or bare except) handler. |
_graph_seam_breach |
_graph_seam_breach(tree: ast.AST) -> str | None |
Return a description of the first raw-graph access in the AST — a neo4j import or a .session(...) call — else None. |
check_graph_seam |
check_graph_seam(py_file: Path) -> str | None |
Return a violation if py_file reaches the graph outside the seam (D13), else None. |
get_layer |
get_layer(module: str) -> int | None |
Return the numeric layer for a module path, or None if unassigned. |
is_protocol_import |
is_protocol_import(imported_module: str) -> bool |
Heuristic: protocol imports live in Layer 0 modules. |
is_internal_init_reexport |
is_internal_init_reexport(module: str, imported_module: str) -> bool |
Allow init.py files to re-export from their own package subtree. |
check_import |
check_import(importing_module: str, imported_module: str) -> str | None |
Return violation description or None if allowed. |
_is_inside_type_checking |
_is_inside_type_checking(node: ast.AST, tree: ast.AST) -> bool |
Check whether an import node is inside an if TYPE_CHECKING: block. |
_is_inside_function_or_class |
_is_inside_function_or_class(node: ast.AST, tree: ast.AST) -> bool |
Check whether an import node is inside a function or class definition. |
_is_inside_def |
_is_inside_def(node: ast.AST, tree: ast.AST) -> bool |
Check whether an import node is def-nested — inside a (sync or async) function body, as opposed to a bare class-body-level import. |
_parse_module |
_parse_module(path: Path) -> tuple[str, ast.AST] | None |
Parse a source file under src/ and return (module_name, ast_tree). |
_iter_ultimate_imports |
_iter_ultimate_imports(tree: ast.AST) -> list[tuple[ast.AST, str]] |
Return (node, imported_module) for every orchestrata_ultimate import node (import x or from x import y) in the tree, unfiltered. |
extract_imports |
extract_imports(path: Path) -> list[tuple[str, str]] |
Return list of (importing_module, imported_module) from a Python file. |
extract_function_scoped_imports |
extract_function_scoped_imports(path: Path) -> list[tuple[str, str]] |
Return (importing_module, imported_module) pairs for def-nested (function-scoped) orchestrata_ultimate imports. |
check_project |
check_project(root: Path, verbose: bool = False) -> tuple[int, list[str], int, int] |
Check all Python files under root. |
main |
main(argv: list[str] | None = None) -> int |
Source
#!/usr/bin/env python3
"""Architecture checker — validates import rules, module boundaries, and layering constraints."""
from __future__ import annotations
import argparse
import ast
import sys
from pathlib import Path
# ── Layer model ──────────────────────────────────────────────────────────────
# Based on architecture-standards.md (the authoritative layer doc — this
# LAYER_ASSIGNMENTS table is the enforcement source; architecture-standards.md
# mirrors it for humans and MUST be kept in sync when this table changes).
# Layer 0 = Vocabulary (no orchestrata imports)
# Layer 1 = Infrastructure
# Layer 2 = Communication
# Layer 3 = Domain Services (no concrete peer imports)
# Layer 4 = Orchestration
# Layer 5 = Interface
LAYER_ASSIGNMENTS: dict[str, int] = {
# Layer 0: Vocabulary
"core.domain": 0,
"core.protocols": 0,
"core.packet": 0,
"core.enums": 0,
"core.interfaces": 0,
"core.base": 0,
"types": 0,
"models": 0,
"interfaces": 0,
"exceptions": 0,
# Layer 1: Infrastructure
"core.config": 1,
"core.logging": 1,
"core.tracing": 1,
"core.monitoring": 1,
"core.patterns": 1,
"core.state": 1,
"core.threading": 1,
"core.adapters": 1,
"core.traceability": 1,
"core.traceability_utils": 1,
"core.redis_factory": 1,
"core.graph_factory": 1,
"core.infra_config": 1,
"core.graph_directives": 1,
"di": 1,
"infra": 1,
# Layer 2: Communication
"cxp": 2,
"postman": 2,
"memory": 2,
# Layer 3: Domain Services
"agents": 3,
"governance": 3,
"knowledge_graph": 3,
"providers": 3,
"verification": 3,
"llm": 3,
"drivers": 3,
"prolog": 3,
# Layer 4: Orchestration
"kernel": 4,
"orchestrator": 4,
"session": 4,
"context": 4,
# Layer 5: Interface
"gateway": 5,
"plugin": 5,
"__main__": 5,
}
# Allowed upward imports per layer.
# Layer N may import from layers <= N.
# Layer 3 has special peer-import rules (see check_peer_import).
UPWARD_IMPORT_LIMITS: dict[int, int] = {
0: 0, # Layer 0 may only import from Layer 0 (but in practice only stdlib)
1: 1,
2: 2,
3: 3,
4: 4,
5: 5,
}
# Known violations that are accepted as legacy / transitional.
# Format: (importing_module_pattern, imported_module_pattern)
KNOWN_VIOLATIONS: set[tuple[str, str]] = {
# Conditional/lazy import for optional Prolog backend (layer-2 module
# safely attempts layer-4 import inside try/except; runtime graceful
# degradation when Prolog worker is unavailable).
(
"orchestrata_ultimate.cxp.prolog_bridge",
"orchestrata_ultimate.orchestrator.prolog_worker",
),
}
# ── D4 composition rule ──────────────────────────────────────────────────────
# Ruled 2026-07-06 (plans/decisions-2026-07-06.md D4, evidence:
# plans/evidence/d4-di-container.md): an adoption audit found 8 of 9 real
# composition roots never touch DIContainer at all — they wire dependencies
# by hand (explicit constructor calls / setters). The container's scope was
# shrunk to the plugin SDK's resolve() touchpoint and the gateway's single
# assembly root (see specs/024-dependency-injection/spec.md § "Descoped
# (2026-07-06)"). This rule keeps that scope from silently regressing: any
# NEW ``DIContainer(...)`` construction outside plugin/gateway/di is flagged.
# Test files are exempt — constructing components directly in tests is the
# established, accepted pattern in this suite, not a violation.
DI_CONTAINER_HOME_PACKAGES: frozenset[str] = frozenset({"plugin", "gateway", "di"})
def _is_test_path(path: Path) -> bool:
"""True if ``path`` lives under a ``tests`` directory or is a test module."""
return "tests" in path.parts or path.name.startswith("test_")
def _di_container_top_level_package(path: Path) -> str | None:
"""Return the top-level ``orchestrata_ultimate.<pkg>`` package name for a
source file path, or None if it isn't under an ``orchestrata_ultimate``
tree at all (e.g. a synthetic fixture rooted directly at a package dir)."""
parts = path.parts
if "orchestrata_ultimate" in parts:
idx = parts.index("orchestrata_ultimate")
return parts[idx + 1] if idx + 1 < len(parts) else None
# Fall back to the first path component (covers fixtures/tests that
# don't nest under a literal "orchestrata_ultimate" directory).
return parts[0] if parts else None
def _instantiates_di_container(tree: ast.AST) -> bool:
"""True if the AST contains a call shaped like ``DIContainer(...)`` or
``some.path.DIContainer(...)``."""
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if isinstance(func, ast.Name) and func.id == "DIContainer":
return True
if isinstance(func, ast.Attribute) and func.attr == "DIContainer":
return True
return False
def check_di_instantiation(py_file: Path) -> str | None:
"""Return a violation description if ``py_file`` instantiates
``DIContainer(...)`` outside the plugin/gateway/di packages, else None.
Independent of the layer-import machinery below: this looks at *calls*,
not imports, and doesn't require a ``src/`` prefix on ``py_file`` (so it
degrades safely if ever pointed at a non-src root).
"""
if _is_test_path(py_file):
return None
if _di_container_top_level_package(py_file) in DI_CONTAINER_HOME_PACKAGES:
return None
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source)
except OSError, SyntaxError, UnicodeDecodeError:
return None
if not _instantiates_di_container(tree):
return None
return (
f"DI composition (D4): {py_file} instantiates DIContainer() outside "
"plugin/gateway/di — composition roots must use explicit "
"constructor/setter wiring (architecture-standards.md §composition)"
)
# Files allowed to construct graph/Redis clients directly: the factories
# themselves (D12 — every other module must go through them).
_DRIVER_FACTORY_FILES = {"graph_factory.py", "redis_factory.py"}
def _constructs_driver_directly(tree: ast.AST) -> str | None:
"""Return a short description of the first direct driver construction in
the AST — ``GraphDatabase.driver(...)`` / ``AsyncGraphDatabase.driver(...)``
or ``redis.Redis(...)`` / ``redis.asyncio.Redis(...)`` — else None."""
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if not isinstance(func, ast.Attribute):
continue
if func.attr == "driver":
base = func.value
name = base.id if isinstance(base, ast.Name) else getattr(base, "attr", "")
if name in ("GraphDatabase", "AsyncGraphDatabase"):
return f"{name}.driver(...) at line {node.lineno}"
if func.attr == "Redis":
base = func.value
name = base.id if isinstance(base, ast.Name) else getattr(base, "attr", "")
if name in ("redis", "asyncio"):
return f"{name}.Redis(...) at line {node.lineno}"
return None
def check_direct_driver_construction(py_file: Path) -> str | None:
"""Return a violation if ``py_file`` constructs a graph/Redis client
directly instead of via the core factories (D12), else None.
Call-shape (AST) based like ``check_di_instantiation`` — string literals
and docstrings never match. Tests and the factory modules are exempt.
"""
if _is_test_path(py_file):
return None
if py_file.name in _DRIVER_FACTORY_FILES:
return None
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source)
except OSError, SyntaxError, UnicodeDecodeError:
return None
hit = _constructs_driver_directly(tree)
if hit is None:
return None
return (
f"Persistence factories (D12): {py_file} constructs {hit} directly — "
"use core.graph_factory.create_graph/create_async_graph or "
"core.redis_factory.create_sync_redis/create_async_redis"
)
# ── D13: operational-graph repository seam (backlog #163 phase 6) ────────────
#
# After the phase 0-5b migration, raw graph access lives ONLY in the seam and
# the documented escape hatches. Two shapes are gated, because phase 5b proved
# imports alone miss attribute-coupled usage (an injected driver never imports
# neo4j — only a `.session(` sweep found QueryMixin, context_injector, etc.):
# (a) `import neo4j` / `from neo4j import ...` module-level or nested;
# (b) `.session(...)` attribute CALLS (AST-shape: strings/docstrings that
# merely mention driver.session() never match).
#
# Allowlist rationale (plans/neo4j-seam-design.md):
# core/graph_factory.py — THE driver factory
# core/graph/repository.py — seam factory/protocol layer
# core/graph/adapters/bolt_sync.py — the seam itself
# core/graph/adapters/bolt_async.py — the seam itself
# knowledge_graph/graph_client.py — plan ruling: "stays as-is — itself
# nearly a repository"
# knowledge_graph/kg_context.py — documented raw-Driver escape hatch
# (external inspection) + typing
# knowledge_graph/semantic_bridge.py — documented escape hatch: sync
# Memgraph client (Bolt) for the
# assembler's synchronous call path;
# the proper semantic_graph adapter
# is async-only, and the Neo4j side
# already goes through
# create_graph_repository()
# session/kg_sync/_base.py — documented escape hatch: bespoke
# deadlock-retry + chunk-splitting
# needs begin_transaction
# semantic_graph/adapters/memgraph.py — the MEMGRAPH adapter: the neo4j
# Bolt package is its client library;
# this file IS that graph's seam
# semantic_graph/ha_controller.py — Memgraph HA controller, same
# factory/promotion/store.py — documented escape hatch: owns the
# ADR-0019 watermark write/read pair
# and exposes its own database-bound
# `session()` to callers (SPEC-033
# B4a/B9). The original rationale —
# "needs raw `Bookmarks` +
# `driver.session(bookmarks=...)`" —
# expired on 2026-07-26: the
# orchgraph engine emits no bookmarks,
# so the watermark is now an explicit
# visibility token and `Bookmarks` is
# gone from this file entirely.
# factory/scheduler/scheduler.py — parks on ADR-0019 watermarks between
# DAG levels. Still allowlisted, but
# for a NARROWER reason than before:
# it no longer builds a bookmarked
# driver session, it borrows the
# promotion store's own
# database-bound `store.session()`.
# D13 matches any `.session(` call
# shape, seam-obtained or not.
_GRAPH_SEAM_ALLOWED: tuple[str, ...] = (
"core/graph_factory.py",
"core/graph/repository.py",
"core/graph/adapters/bolt_sync.py",
"core/graph/adapters/bolt_async.py",
"knowledge_graph/graph_client.py",
"knowledge_graph/kg_context.py",
"knowledge_graph/semantic_bridge.py",
"session/kg_sync/_base.py",
"semantic_graph/adapters/memgraph.py",
"semantic_graph/ha_controller.py",
"factory/promotion/store.py",
"factory/scheduler/scheduler.py",
)
def _import_probe_lines(tree: ast.AST) -> set[int]:
"""Line numbers inside ``try:`` bodies that have an ``ImportError`` (or
``ModuleNotFoundError``, or bare ``except``) handler. A bare
``import neo4j`` there is an availability PROBE — it feature-gates on the
optional package's presence and reaches nothing — so D13 exempts it.
``from neo4j import X`` stays gated even here: it binds driver symbols."""
probe_lines: set[int] = set()
for node in ast.walk(tree):
if not isinstance(node, ast.Try):
continue
catches_import_error = False
for handler in node.handlers:
if handler.type is None:
catches_import_error = True
break
names = []
for t in (
handler.type.elts
if isinstance(handler.type, ast.Tuple)
else [handler.type]
):
if isinstance(t, ast.Name):
names.append(t.id)
if "ImportError" in names or "ModuleNotFoundError" in names:
catches_import_error = True
break
if not catches_import_error:
continue
for stmt in node.body:
end = getattr(stmt, "end_lineno", stmt.lineno)
probe_lines.update(range(stmt.lineno, end + 1))
return probe_lines
def _graph_seam_breach(tree: ast.AST) -> str | None:
"""Return a description of the first raw-graph access in the AST —
a ``neo4j`` import or a ``.session(...)`` call — else None."""
probe_lines = _import_probe_lines(tree)
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name == "neo4j" or alias.name.startswith("neo4j."):
if node.lineno in probe_lines:
continue # availability probe, not graph access
return f"import {alias.name} at line {node.lineno}"
elif isinstance(node, ast.ImportFrom):
mod = node.module or ""
if node.level == 0 and (mod == "neo4j" or mod.startswith("neo4j.")):
return f"from {mod} import ... at line {node.lineno}"
elif isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Attribute) and func.attr == "session":
return f".session(...) call at line {node.lineno}"
return None
def check_graph_seam(py_file: Path) -> str | None:
"""Return a violation if ``py_file`` reaches the graph outside the seam
(D13), else None. Tests and the allowlisted seam/escape-hatch files are
exempt."""
if _is_test_path(py_file):
return None
posix = py_file.as_posix()
if any(posix.endswith(allowed) for allowed in _GRAPH_SEAM_ALLOWED):
return None
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source)
except OSError, SyntaxError, UnicodeDecodeError:
return None
hit = _graph_seam_breach(tree)
if hit is None:
return None
return (
f"Graph seam (D13): {py_file} reaches the graph driver raw ({hit}) — go through "
"core.graph (GraphRepository / create_graph_repository) or add the "
"file to the documented escape-hatch allowlist with a reason "
"(plans/neo4j-seam-design.md phase 6)"
)
def get_layer(module: str) -> int | None:
"""Return the numeric layer for a module path, or None if unassigned."""
# Strip package prefix so we look up relative to the package root.
prefix = "orchestrata_ultimate."
if module.startswith(prefix):
module = module[len(prefix) :]
parts = module.split(".")
# Try progressively longer prefixes
for i in range(len(parts), 0, -1):
candidate = ".".join(parts[:i])
if candidate in LAYER_ASSIGNMENTS:
return LAYER_ASSIGNMENTS[candidate]
return None
def is_protocol_import(imported_module: str) -> bool:
"""Heuristic: protocol imports live in Layer 0 modules."""
layer = get_layer(imported_module)
return layer == 0
def is_internal_init_reexport(module: str, imported_module: str) -> bool:
"""Allow __init__.py files to re-export from their own package subtree."""
if not module.endswith(".__init__"):
return False
pkg = module.removesuffix(".__init__") + "."
return imported_module.startswith(pkg)
def check_import(importing_module: str, imported_module: str) -> str | None:
"""Return violation description or None if allowed."""
if not imported_module.startswith("orchestrata_ultimate"):
return None
importer_layer = get_layer(importing_module)
imported_layer = get_layer(imported_module)
if importer_layer is None or imported_layer is None:
return None # Unassigned modules are not checked
# Allow __init__ re-exports
if is_internal_init_reexport(importing_module, imported_module):
return None
# Upward import check
if imported_layer > importer_layer:
return (
f"upward import: {importing_module} (layer {importer_layer}) "
f"imports {imported_module} (layer {imported_layer})"
)
# (A former "layer-0 violation" branch here was unreachable: any
# imported_layer > importer_layer already returned as an upward import
# above, so a layer-0 importer could only reach it with imported_layer == 0,
# which is allowed. Removed 2026-07-06.)
# Layer 3 peer-import check: concrete imports from *different* peer Layer-3
# top-level packages are forbidden. Imports within the same package tree
# (e.g. agents.foo -> agents.bar) are allowed.
# Protocol imports (Layer 0) are always fine.
if importer_layer == 3 and imported_layer == 3:
if not is_protocol_import(imported_module):
# Determine top-level package for each (e.g. "agents", "knowledge_graph")
importer_pkg = importing_module.removeprefix("orchestrata_ultimate.").split(
"."
)[0]
imported_pkg = imported_module.removeprefix("orchestrata_ultimate.").split(
"."
)[0]
if importer_pkg != imported_pkg:
return (
f"peer import: {importing_module} (layer 3) "
f"imports concrete {imported_module} (layer 3); "
f"use protocols instead"
)
return None
def _is_inside_type_checking(node: ast.AST, tree: ast.AST) -> bool:
"""Check whether an import node is inside an ``if TYPE_CHECKING:`` block."""
for parent in ast.walk(tree):
if isinstance(parent, ast.If):
if isinstance(parent.test, ast.Name) and parent.test.id == "TYPE_CHECKING":
for child in ast.walk(parent):
if child is node:
return True
return False
def _is_inside_function_or_class(node: ast.AST, tree: ast.AST) -> bool:
"""Check whether an import node is inside a function or class definition."""
for parent in ast.walk(tree):
if isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
for child in ast.walk(parent):
if child is node:
return True
return False
def _is_inside_def(node: ast.AST, tree: ast.AST) -> bool:
"""Check whether an import node is def-nested — inside a (sync or async)
function body, as opposed to a bare class-body-level import. This is the
"function-scoped" / lazy-import shape (RULED D7): imports guarded behind
a call, used for optional/heavy deps or to sidestep a layer exception."""
for parent in ast.walk(tree):
if isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef)):
for child in ast.walk(parent):
if child is node:
return True
return False
def _parse_module(path: Path) -> tuple[str, ast.AST] | None:
"""Parse a source file under ``src/`` and return (module_name, ast_tree).
Returns None if the module name can't be derived, the file can't be
read, or it fails to parse.
"""
rel = path.relative_to("src")
module = str(rel.with_suffix("")).replace("/", ".").replace("\\", ".")
try:
source = path.read_text(encoding="utf-8")
except Exception:
return None
try:
tree = ast.parse(source)
except SyntaxError:
return None
return module, tree
def _iter_ultimate_imports(tree: ast.AST) -> list[tuple[ast.AST, str]]:
"""Return (node, imported_module) for every orchestrata_ultimate import
node (``import x`` or ``from x import y``) in the tree, unfiltered."""
found: list[tuple[ast.AST, str]] = []
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
if node.module and node.module.startswith("orchestrata_ultimate"):
found.append((node, node.module))
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.startswith("orchestrata_ultimate"):
found.append((node, alias.name))
return found
def extract_imports(path: Path) -> list[tuple[str, str]]:
"""Return list of (importing_module, imported_module) from a Python file.
Excludes ``TYPE_CHECKING``-guarded imports and any import nested inside a
function or class body — those are the accepted lazy-import shape, and
are surfaced separately by :func:`extract_function_scoped_imports`.
"""
imports: list[tuple[str, str]] = []
parsed = _parse_module(path)
if parsed is None:
return imports
module, tree = parsed
for node, imported_module in _iter_ultimate_imports(tree):
if _is_inside_type_checking(node, tree):
continue
if _is_inside_function_or_class(node, tree):
continue
imports.append((module, imported_module))
return imports
def extract_function_scoped_imports(path: Path) -> list[tuple[str, str]]:
"""Return (importing_module, imported_module) pairs for def-nested
(function-scoped) orchestrata_ultimate imports.
This is the accepted lazy-import pattern (RULED D7 — see
architecture-standards.md): function-scoped cross-layer imports for
optional/heavy deps or layer-exception avoidance. They are excluded from
the main violation check but reported at INFO level for visibility, since
they still create real runtime coupling.
"""
imports: list[tuple[str, str]] = []
parsed = _parse_module(path)
if parsed is None:
return imports
module, tree = parsed
for node, imported_module in _iter_ultimate_imports(tree):
if _is_inside_type_checking(node, tree):
continue
if _is_inside_def(node, tree):
imports.append((module, imported_module))
return imports
def check_project(root: Path, verbose: bool = False) -> tuple[int, list[str], int, int]:
"""Check all Python files under root.
Returns (violation_count, messages, files_checked,
function_scoped_cross_layer_count). The last figure is an INFO-level
count of def-nested imports that reach a higher layer (RULED D7) — it
never affects violation_count or the process exit code.
"""
violations: list[str] = []
files_checked = 0
function_scoped_cross_layer_count = 0
for py_file in sorted(root.rglob("*.py")):
if "__pycache__" in py_file.parts:
continue
files_checked += 1
for importing_module, imported_module in extract_imports(py_file):
violation = check_import(importing_module, imported_module)
if violation:
# Check against known waivers
waived = any(
importing_module.startswith(imp_pat)
and imported_module.startswith(exp_pat)
for imp_pat, exp_pat in KNOWN_VIOLATIONS
)
if not waived:
msg = f"{importing_module} -> {imported_module}: {violation}"
violations.append(msg)
if verbose:
print(f" VIOLATION: {msg}")
for importing_module, imported_module in extract_function_scoped_imports(
py_file
):
# Count every def-nested import that WOULD be a violation (upward
# OR layer-3 peer) if it were eager — the full set of lazy imports
# dodging the layer rules, per D7's visibility intent.
if check_import(importing_module, imported_module) is not None:
function_scoped_cross_layer_count += 1
di_violation = check_di_instantiation(py_file)
if di_violation is not None:
violations.append(di_violation)
if verbose:
print(f" VIOLATION: {di_violation}")
driver_violation = check_direct_driver_construction(py_file)
if driver_violation is not None:
violations.append(driver_violation)
if verbose:
print(f" VIOLATION: {driver_violation}")
seam_violation = check_graph_seam(py_file)
if seam_violation is not None:
violations.append(seam_violation)
if verbose:
print(f" VIOLATION: {seam_violation}")
return (
len(violations),
violations,
files_checked,
function_scoped_cross_layer_count,
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Validate architecture import rules")
parser.add_argument(
"path", nargs="?", default="src/orchestrata_ultimate", help="Path to check"
)
parser.add_argument(
"-v", "--verbose", action="store_true", help="Show per-violation details"
)
parser.add_argument("-o", "--output", help="Write report to file")
parser.add_argument(
"--strict", action="store_true", help="Exit non-zero on any violation"
)
args = parser.parse_args(argv)
root = Path(args.path)
if not root.exists():
print(f"ERROR: path does not exist: {root}", file=sys.stderr)
return 1
print(f"Checking architecture rules under {root.resolve()} ...")
violation_count, violations, files_checked, function_scoped_cross_layer_count = (
check_project(root, verbose=args.verbose)
)
lines: list[str] = []
lines.append(f"Files checked: {files_checked}")
lines.append(f"Violations found: {violation_count}")
lines.append(
f"INFO: {function_scoped_cross_layer_count} function-scoped cross-layer "
"imports (accepted pattern, see architecture-standards.md)"
)
if violations:
lines.append("")
lines.append("Violations:")
lines.extend(f" - {v}" for v in violations)
report = "\n".join(lines)
print(report)
if args.output:
Path(args.output).write_text(report + "\n", encoding="utf-8")
print(f"\nReport written to {args.output}")
# In non-strict mode we exit 0 so the gate passes while violations are
# documented. In strict mode any violation is a failure.
if args.strict and violation_count > 0:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())