orchestrata_ultimate.orchestrator.impact.dependency_analyzer

Reference › orchestrata_ultimate › orchestrator › impact › dependency_analyzer

The graph-backed part of impact analysis: it asks the knowledge graph what depends on a package or module. analyzer is its caller. It uses GraphClient and GraphClientConfig from the knowledge_graph package; with no configuration passed, GraphClientConfig() resolves the graph connection settings itself. See the knowledge graph module.

The client is created lazily by _ensure_client on first use and released by close. find_affected_scope takes an optional target package, module and file and returns a dictionary with affected packages, modules, files, upstream, downstream and tests. For a package it calls _find_by_package, which uses _find_downstream and _find_upstream; for a module it uses _find_by_module; tests come from _find_tests. All of these run Cypher through _query. Two further methods serve contract checks: find_interface returns a module's interface and find_circular_deps returns dependency cycles through a package.

At a glance

Class diagram

Dependency diagram

API

Classes

DependencyAnalyzer

Decorators: traces(spec='019', story='US-002', reqs=['REQ-019-002']) · Line 19

Analyzes dependency graph from knowledge graph.

Source

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

# mypy: ignore-errors
"""Dependency graph analyzer - queries the graph for impact analysis.

Analyzes the knowledge graph to find affected packages, modules,
files, and their dependencies based on IMPORTS and CALLS relationships.
"""

import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.config import GraphClientConfig
from orchestrata_ultimate.knowledge_graph.exceptions import KnowledgeGraphError
from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

logger = logging.getLogger(__name__)


@traces(spec="019", story="US-002", reqs=["REQ-019-002"])
class DependencyAnalyzer:
    """Analyzes dependency graph from knowledge graph.

    Queries the graph to find affected packages, modules, files,
    and their upstream/downstream dependencies.
    """

    def __init__(self, config: GraphClientConfig | None = None) -> None:
        """Initialize analyzer with graph config.

        Args:
            config: Graph connection configuration. Uses
                environment defaults if not provided.
        """
        # GraphClientConfig() resolves uri/database/username/password at
        # construction time (core.graph_factory.resolve_uri /
        # resolve_kg_database — RUL-003, no literal fallback). This used to
        # pin uri=NEO4J_URI / username=GRAPH_USERNAME / password=GRAPH_PASSWORD
        # explicitly, but those were exactly GraphClientConfig()'s own
        # defaults — the override was redundant and, worse, baked in the
        # import-time NEO4J_URI snapshot instead of resolving per-construction.
        self._config = config or GraphClientConfig()
        self._client: GraphClient | None = None

    async def _ensure_client(self) -> GraphClient:
        """Ensure graph client is connected.

        Returns:
            Connected GraphClient instance.
        """
        if self._client is None:
            self._client = GraphClient(config=self._config)
            await self._client.connect()
        return self._client

    async def find_affected_scope(
        self,
        target_package: str | None = None,
        target_module: str | None = None,
        target_file: str | None = None,
    ) -> dict:
        """Find all affected packages, modules, and files.

        Args:
            target_package: Package name to analyze.
            target_module: Module name to analyze.
            target_file: File path to analyze.

        Returns:
            Dict with keys: affected_packages, affected_modules,
            affected_files, upstream, downstream, tests.
        """
        await self._ensure_client()

        result = {
            "affected_packages": [],
            "affected_modules": [],
            "affected_files": [],
            "upstream": [],
            "downstream": [],
            "tests": [],
        }

        if target_package:
            pkg_result = await self._find_by_package(target_package)
            result.update(pkg_result)

        if target_module:
            module_files = await self._find_by_module(target_module)
            result["affected_files"] = module_files

        result["tests"] = await self._find_tests(target_file)
        return result

    async def _find_by_package(self, package: str) -> dict:
        """Find modules, files, and dependencies for a package.

        Args:
            package: Package name to query.

        Returns:
            Dict with affected_modules, affected_files,
            downstream, and upstream lists.
        """
        modules = await self._query(
            "MATCH (p:Package {name: $package})-[:HAS_MODULE]->(m:Module) "
            "RETURN m.name as name, m.path as path",
            {"package": package},
        )
        files = await self._query(
            "MATCH (p:Package {name: $package})-[:HAS_MODULE]->"
            "(m:Module)-[:HAS_FILE]->(f:File) "
            "RETURN f.path as path, f.name as name",
            {"package": package},
        )
        downstream = await self._find_downstream(package)
        upstream = await self._find_upstream(package)

        return {
            "affected_modules": [
                {"name": m["name"], "path": m["path"]} for m in modules
            ],
            "affected_files": [{"name": f["name"], "path": f["path"]} for f in files],
            "downstream": [d["package"] for d in downstream],
            "upstream": [u["package"] for u in upstream],
        }

    async def _find_downstream(self, package: str) -> list[dict]:
        """Find packages that this package imports.

        Args:
            package: Source package name.

        Returns:
            List of dicts with 'package' key.
        """
        return await self._query(
            "MATCH (p:Package {name: $package})-[:HAS_MODULE]->"
            "(m:Module)-[:HAS_FILE]->(f:File) "
            "MATCH (f)-[:IMPORTS]->(dep:File) "
            "MATCH (dep)-[:HAS_FILE]->(dm:Module)"
            "-[:HAS_PACKAGE]->(dp:Package) "
            "WHERE dp.name <> $package "
            "RETURN DISTINCT dp.name as package",
            {"package": package},
        )

    async def _find_upstream(self, package: str) -> list[dict]:
        """Find packages that import this package.

        Args:
            package: Target package name.

        Returns:
            List of dicts with 'package' key.
        """
        return await self._query(
            "MATCH (p:Package {name: $package})-[:HAS_MODULE]->"
            "(m:Module)-[:HAS_FILE]->(f:File) "
            "MATCH (other:File)-[:IMPORTS]->(f) "
            "MATCH (other)-[:HAS_FILE]->(om:Module)"
            "-[:HAS_PACKAGE]->(op:Package) "
            "WHERE op.name <> $package "
            "RETURN DISTINCT op.name as package",
            {"package": package},
        )

    async def _find_by_module(self, module: str) -> list[dict]:
        """Find files belonging to a module.

        Args:
            module: Module name to query.

        Returns:
            List of dicts with file paths.
        """
        files = await self._query(
            "MATCH (m:Module {name: $module})-[:HAS_FILE]->(f:File) "
            "RETURN f.path as path",
            {"module": module},
        )
        return [{"path": f["path"]} for f in files]

    async def _find_tests(self, target_file: str | None = None) -> list[dict]:
        """Find test files, optionally filtered by target.

        Args:
            target_file: If provided, only return tests that
                import this file.

        Returns:
            List of dicts with test file paths and names.
        """
        tests = await self._query(
            "MATCH (test:File)-[:DEFINES]->(tc:Class) "
            "WHERE tc.name CONTAINS 'Test' OR tc.name ENDS WITH 'Test' "
            "OPTIONAL MATCH (test)-[:IMPORTS]->(target:File) "
            "WHERE target.path = $target_file OR $target_file IS NULL "
            "RETURN DISTINCT test.path as path, test.name as name",
            {"target_file": target_file},
        )
        return [{"path": t["path"], "name": t["name"]} for t in tests]

    async def find_interface(self, module_name: str) -> dict:
        """Find the interface contract for a module.

        Args:
            module_name: Name of the module to inspect.

        Returns:
            Dict with module name, classes, and functions.
        """
        await self._ensure_client()

        classes = await self._query(
            "MATCH (m:Module {name: $module})-[:HAS_FILE]->"
            "(f:File)-[:DEFINES]->(c:Class) "
            "RETURN c.name as name, c.line_start as line",
            {"module": module_name},
        )
        functions = await self._query(
            "MATCH (m:Module {name: $module})-[:HAS_FILE]->"
            "(f:File)-[:DEFINES]->(fn:Function) "
            "RETURN fn.name as name, fn.signature as signature, "
            "fn.is_async as is_async",
            {"module": module_name},
        )

        return {
            "module": module_name,
            "classes": [{"name": c["name"], "line": c["line"]} for c in classes],
            "functions": [
                {
                    "name": f["name"],
                    "signature": f.get("signature"),
                    "is_async": f.get("is_async", False),
                }
                for f in functions
            ],
        }

    async def find_circular_deps(self, package: str) -> list[list[str]]:
        """Find circular dependencies involving a package.

        Uses a Cypher path query to detect cycles in the
        IMPORTS relationship graph within a depth of 5 hops.

        Args:
            package: Package name to check for cycles.

        Returns:
            List of cycles, each cycle is a list of file paths.
        """
        await self._ensure_client()

        cycles = await self._query(
            "MATCH (p:Package {name: $package})-[:HAS_MODULE]->"
            "(m:Module)-[:HAS_FILE]->(start:File) "
            "MATCH path = (start)-[:IMPORTS*2..5]->(start) "
            "RETURN [n IN nodes(path) | n.path] AS cycle "
            "LIMIT 10",
            {"package": package},
        )
        return [c["cycle"] for c in cycles]

    async def _query(self, cypher: str, params: dict) -> list[dict]:
        """Execute a Cypher query with error handling.

        Args:
            cypher: Cypher query string.
            params: Query parameters.

        Returns:
            List of record dicts, empty on error.
        """
        try:
            # `_ensure_client()` returns the connected client (guaranteed
            # non-None); every caller already invokes it before `_query`,
            # so this is a cheap no-op reuse of `self._client` at runtime,
            # not a new connection.
            client = await self._ensure_client()
            result = await client.query(cypher, params)
            return [dict(record) for record in result] if result else []
        except ConnectionError as exc:
            logger.error("Graph connection error: %s", exc)
            return []
        except RuntimeError as exc:
            logger.error("Query runtime error: %s", exc)
            return []
        except KnowledgeGraphError as exc:
            logger.error("Knowledge graph error: %s", exc)
            return []

    async def close(self) -> None:
        """Close the graph client and release resources."""
        if self._client:
            await self._client.close()
            self._client = None