orchestrata_tools.dependency.import_graph

Reference › orchestrata_tools › dependency › import_graph

This module builds the internal import graph of an ASTCodebase. ImportGraph scans every module's imports and keeps only those whose target starts with orchestrata_ultimate, storing forward edges and reverse edges. It answers dependencies_of and dependents_of for a module, all_modules, fan_in and fan_out counts, and most_depended_on, the modules with the highest fan-in. cycle_detector reads the graph to find cycles. The models come from ast_models.

At a glance

Class diagram

Dependency diagram

API

Classes

ImportGraph

Line 12

Build and query the internal import dependency graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_tools/dependency/import_graph.py

"""Internal dependency graph builder."""

from __future__ import annotations

from collections import defaultdict
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from orchestrata_tools.models.ast_models import ASTCodebase


class ImportGraph:
    """Build and query the internal import dependency graph."""

    def __init__(self, codebase: ASTCodebase):
        self.codebase = codebase
        self._edges: dict[str, set[str]] = defaultdict(set)
        self._reverse_edges: dict[str, set[str]] = defaultdict(set)
        self._build()

    def _build(self) -> None:
        for module in self.codebase.modules:
            src = module.path
            for imp in module.imports:
                target = imp.module
                if target.startswith("orchestrata_ultimate"):
                    self._edges[src].add(target)
                    self._reverse_edges[target].add(src)

    def dependencies_of(self, module_path: str) -> set[str]:
        """Return modules that `module_path` imports."""
        return self._edges.get(module_path, set())

    def dependents_of(self, module_path: str) -> set[str]:
        """Return modules that import `module_path`."""
        return self._reverse_edges.get(module_path, set())

    def all_modules(self) -> set[str]:
        """Return all module paths in the graph."""
        return set(self._edges) | set(self._reverse_edges)

    def fan_in(self, module_path: str) -> int:
        """Number of modules that depend on this module."""
        return len(self.dependents_of(module_path))

    def fan_out(self, module_path: str) -> int:
        """Number of modules this module depends on."""
        return len(self.dependencies_of(module_path))

    def most_depended_on(self, n: int = 10) -> list[tuple[str, int]]:
        """Return modules with highest fan-in."""
        modules = self.all_modules()
        scored = [(m, self.fan_in(m)) for m in modules]
        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:n]