orchestrata_ultimate.knowledge_graph.dependency_analyzer

Reference › orchestrata_ultimate › knowledge_graph › dependency_analyzer

DependencyAnalyzer builds import-level dependency graphs from file analyses produced by ast_analyzer and code_parser. analyze_imports takes a list of FileAnalysis, builds a map from module names to file paths (_build_module_map), resolves each import against it (_resolve_module) and returns a DependencyGraph of Dependency edges typed with RelationshipType. Circular dependencies are detected by a depth-first search in _detect_cycles. Further queries on a graph are get_transitive_deps (bounded by max_depth, default 10), get_reverse_deps (which files depend on a given one) and get_dependency_layers, a topological layering. All work happens in memory on the supplied analyses; nothing is read from or written to the graph database. Only tests import this module.

At a glance

Class diagram

Dependency diagram

API

Classes

DependencyAnalyzer

Line 20

Analyzes dependencies between files and modules.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/dependency_analyzer.py

"""Dependency analysis for the knowledge graph.

Analyzes import statements to build dependency graphs,
detect circular dependencies, and compute transitive dependencies.
"""

import logging
from collections import defaultdict

from orchestrata_ultimate.knowledge_graph.types import (
    Dependency,
    DependencyGraph,
    FileAnalysis,
    RelationshipType,
)

logger = logging.getLogger(__name__)


class DependencyAnalyzer:
    """Analyzes dependencies between files and modules."""

    def __init__(self) -> None:
        self._module_to_file: dict[str, str] = {}

    def analyze_imports(self, analyses: list[FileAnalysis]) -> DependencyGraph:
        """Build a dependency graph from file analyses."""
        # Build module-to-file mapping
        self._build_module_map(analyses)

        nodes: set[str] = set()
        edges: list[Dependency] = []
        adjacency: dict[str, set[str]] = defaultdict(set)

        for analysis in analyses:
            source = analysis.file_path
            nodes.add(source)

            for imp in analysis.imports:
                target = self._resolve_module(imp.target_module)
                if target and target != source:
                    nodes.add(target)
                    if target not in adjacency[source]:
                        adjacency[source].add(target)
                        edges.append(
                            Dependency(
                                source_id=source,
                                target_id=target,
                                dependency_type=RelationshipType.IMPORTS,
                                metadata={"imported_names": imp.imported_names},
                            )
                        )

        # Detect circular dependencies
        cycles = self._detect_cycles(adjacency)

        return DependencyGraph(
            nodes=sorted(nodes),
            edges=edges,
            circular_dependencies=cycles,
        )

    def get_transitive_deps(
        self, file_path: str, graph: DependencyGraph, max_depth: int = 10
    ) -> list[str]:
        """Get all transitive dependencies of a file."""
        adjacency = self._build_adjacency(graph)
        visited: set[str] = set()
        result: list[str] = []

        def dfs(node: str, depth: int) -> None:
            if depth > max_depth or node in visited:
                return
            visited.add(node)
            for neighbor in adjacency.get(node, set()):
                if neighbor not in visited:
                    result.append(neighbor)
                    dfs(neighbor, depth + 1)

        dfs(file_path, 0)
        return result

    def get_reverse_deps(self, file_path: str, graph: DependencyGraph) -> list[str]:
        """Get all files that depend on the given file."""
        reverse: dict[str, set[str]] = defaultdict(set)
        for edge in graph.edges:
            reverse[edge.target_id].add(edge.source_id)

        visited: set[str] = set()
        result: list[str] = []

        def dfs(node: str) -> None:
            if node in visited:
                return
            visited.add(node)
            for dep in reverse.get(node, set()):
                if dep not in visited:
                    result.append(dep)
                    dfs(dep)

        dfs(file_path)
        return result

    def get_dependency_layers(self, graph: DependencyGraph) -> list[list[str]]:
        """Compute dependency layers (topological sort layers)."""
        adjacency = self._build_adjacency(graph)
        in_degree: dict[str, int] = defaultdict(int)

        for node in graph.nodes:
            if node not in in_degree:
                in_degree[node] = 0

        for edge in graph.edges:
            in_degree[edge.target_id] += 1

        layers: list[list[str]] = []
        remaining = set(graph.nodes)

        while remaining:
            layer = [n for n in remaining if in_degree.get(n, 0) == 0]
            if not layer:
                # Cycle -- add all remaining as final layer
                layers.append(sorted(remaining))
                break
            layers.append(sorted(layer))
            for node in layer:
                remaining.discard(node)
                for neighbor in adjacency.get(node, set()):
                    in_degree[neighbor] -= 1

        return layers

    def _build_module_map(self, analyses: list[FileAnalysis]) -> None:
        """Build a mapping from module name to file path."""
        for analysis in analyses:
            path = analysis.file_path
            # Convert file path to module name
            module = path.replace("/", ".").replace("\\", ".")
            if module.endswith(".py"):
                module = module[:-3]
            # Store multiple forms
            self._module_to_file[module] = path
            # Also store the last component
            parts = module.split(".")
            if parts:
                self._module_to_file[parts[-1]] = path

    def _resolve_module(self, module_name: str) -> str | None:
        """Resolve a module name to a file path."""
        if module_name in self._module_to_file:
            return self._module_to_file[module_name]
        # Try prefix matches
        for mod, path in self._module_to_file.items():
            if mod.endswith(f".{module_name}"):
                return path
        return None

    def _detect_cycles(self, adjacency: dict[str, set[str]]) -> list[list[str]]:
        """Detect circular dependencies using DFS."""
        cycles: list[list[str]] = []
        visited: set[str] = set()
        rec_stack: set[str] = set()
        path: list[str] = []

        def dfs(node: str) -> None:
            visited.add(node)
            rec_stack.add(node)
            path.append(node)

            for neighbor in adjacency.get(node, set()):
                if neighbor not in visited:
                    dfs(neighbor)
                elif neighbor in rec_stack:
                    # Found a cycle
                    cycle_start = path.index(neighbor)
                    cycle = path[cycle_start:] + [neighbor]
                    cycles.append(cycle)

            path.pop()
            rec_stack.discard(node)

        for node in adjacency:
            if node not in visited:
                dfs(node)

        return cycles

    def _build_adjacency(self, graph: DependencyGraph) -> dict[str, set[str]]:
        """Build adjacency map from graph edges."""
        adj: dict[str, set[str]] = defaultdict(set)
        for edge in graph.edges:
            adj[edge.source_id].add(edge.target_id)
        return adj