orchestrata_ultimate.knowledge_graph.graph_node_mixins

Reference › orchestrata_ultimate › knowledge_graph › graph_node_mixins

Two Protocol mixins that carry the typed-node and file-oriented operations of graph_client, split out to keep the client under the project's size limit while callers still write client.create_class_node(...) through inheritance. Each declares _execute, which GraphClient provides. GraphNodeCreationMixin offers create_package_node, create_module_node, create_class_node, create_function_node and, for Prolog sources (REQ-008-018), create_predicate_node and create_rule_node; each creates or updates the node using the conversion functions in nodes and returns its id. GraphFileQueryMixin offers get_project_files, get_file_entities (what a File defines), delete_file_cascade (remove a file and all child entities, returning the count) and update_file_line_count, each scoped by project id. All of them run Cypher against the operational graph (orchkg on orchgraph).

At a glance

Class diagram

Dependency diagram

API

Classes

GraphNodeCreationMixin (protocol)

Bases: Protocol · Line 28

Typed-node creation helpers for GraphClient.

GraphFileQueryMixin (protocol)

Bases: Protocol · Line 141

File-query and file-mutation helpers for GraphClient.

Source

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

# mypy: ignore-errors
"""Mixin classes for GraphClient node operations.

Splits typed-node creation and file-query methods out of the main client
so the client stays under the god-class line limit while callers continue
to use the same ``client.create_class_node(...)`` interface via inheritance.
"""

from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.nodes import (
    class_node_to_cypher,
    function_node_to_cypher,
    predicate_node_to_cypher,
    rule_node_to_cypher,
)
from orchestrata_ultimate.knowledge_graph.types import (
    ClassNode,
    FunctionNode,
    ModuleNode,
    PackageNode,
    PredicateNode,
    RuleNode,
)


class GraphNodeCreationMixin(Protocol):
    """Typed-node creation helpers for GraphClient.

    All methods delegate to ``self._execute`` which is provided by the
    concrete ``GraphClient`` class.
    """

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-021"])
    async def _execute(
        self, mode: str, cypher: str, params: dict[str, Any]
    ) -> list[dict[str, Any]]:
        """Execute a Cypher statement. Provided by GraphClient."""
        ...

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-021"])
    async def create_package_node(self, package_node: PackageNode) -> str:
        """Create or update a Package node.

        Args:
            package_node: The PackageNode to persist.

        Returns:
            The node ID.
        """
        label = "Package"
        props = package_node.to_dict()
        cypher = f"MERGE (n:{label} {{name: $name}}) SET n = $props RETURN n.id AS id"
        params = {"name": package_node.name, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else package_node.id

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-021"])
    async def create_module_node(self, module_node: ModuleNode) -> str:
        """Create or update a Module node.

        Args:
            module_node: The ModuleNode to persist.

        Returns:
            The node ID.
        """
        label = "Module"
        props = module_node.to_dict()
        cypher = f"MERGE (n:{label} {{path: $path}}) SET n = $props RETURN n.id AS id"
        params = {"path": module_node.path, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else module_node.id

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-021"])
    async def create_class_node(self, class_node: ClassNode) -> str:
        """Create or update a Class node.

        Args:
            class_node: The ClassNode to persist.

        Returns:
            The node ID.
        """
        label, props = class_node_to_cypher(class_node)
        cypher = f"MERGE (n:{label} {{id: $id}}) SET n = $props RETURN n.id AS id"
        params = {"id": class_node.id, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else class_node.id

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-021"])
    async def create_function_node(self, func_node: FunctionNode) -> str:
        """Create or update a Function node.

        Args:
            func_node: The FunctionNode to persist.

        Returns:
            The node ID.
        """
        label, props = function_node_to_cypher(func_node)
        cypher = f"MERGE (n:{label} {{id: $id}}) SET n = $props RETURN n.id AS id"
        params = {"id": func_node.id, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else func_node.id

    @traces(spec="008", story="US-001", reqs=["REQ-008-018"])
    async def create_predicate_node(self, pred_node: PredicateNode) -> str:
        """Create or update a Predicate node (Prolog, REQ-008-018).

        Args:
            pred_node: The PredicateNode to persist.

        Returns:
            The node ID.
        """
        label, props = predicate_node_to_cypher(pred_node)
        cypher = f"MERGE (n:{label} {{id: $id}}) SET n = $props RETURN n.id AS id"
        params = {"id": pred_node.id, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else pred_node.id

    @traces(spec="008", story="US-001", reqs=["REQ-008-018"])
    async def create_rule_node(self, rule_node: RuleNode) -> str:
        """Create or update a Rule node (Prolog, REQ-008-018).

        Args:
            rule_node: The RuleNode to persist.

        Returns:
            The node ID.
        """
        label, props = rule_node_to_cypher(rule_node)
        cypher = f"MERGE (n:{label} {{id: $id}}) SET n = $props RETURN n.id AS id"
        params = {"id": rule_node.id, "props": props}
        result = await self._execute("write", cypher, params)
        return result[0]["id"] if result else rule_node.id


class GraphFileQueryMixin(Protocol):
    """File-query and file-mutation helpers for GraphClient.

    All methods delegate to ``self._execute`` which is provided by the
    concrete ``GraphClient`` class.
    """

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def _execute(
        self, mode: str, cypher: str, params: dict[str, Any]
    ) -> list[dict[str, Any]]:
        """Execute a Cypher statement. Provided by GraphClient."""
        ...

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def get_project_files(self, project_id: str) -> list[dict[str, Any]]:
        """Return all File nodes for a project.

        Uses SKIP/LIMIT pagination (500 per batch) to handle
        large graphs without memory pressure.

        Args:
            project_id: The project node ID to query files for.

        Returns:
            List of dicts with keys: path, hash, line_count, id.
        """
        all_files: list[dict[str, Any]] = []
        skip = 0
        batch_size = 500
        while True:
            cypher = (
                "MATCH (p:Project {id: $project_id}) "
                "-[:HAS_FILE]-> (f:File) "
                "RETURN f.path AS path, "
                "f.hash AS hash, "
                "f.line_count AS line_count, "
                "f.id AS id "
                "ORDER BY f.path "
                "SKIP $skip LIMIT $limit"
            )
            params = {"project_id": project_id, "skip": skip, "limit": batch_size}
            batch = await self._execute("read", cypher, params)
            if not batch:
                break
            all_files.extend(batch)
            if len(batch) < batch_size:
                break
            skip += batch_size
        return all_files

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-079"])
    async def get_file_entities(
        self, file_path: str, project_id: str
    ) -> dict[str, list[dict[str, Any]]]:
        """Return entities defined by a File node.

        Queries all Class, Function, and Predicate nodes
        connected via DEFINES relationships from the File.

        Args:
            file_path: The file path to look up.
            project_id: The owning project's id — File.path is
                project-relative and collides across projects (2026-07-12),
                so the match must be scoped to avoid reading another
                project's entities.

        Returns:
            Dict with keys 'classes', 'functions', 'predicates',
            each a list of dicts with 'name' and 'id'.
        """
        result: dict[str, list[dict[str, Any]]] = {
            "classes": [],
            "functions": [],
            "predicates": [],
        }
        cypher = (
            "MATCH (f:File {path: $path, project_id: $project_id})"
            "-[:DEFINES]->(e) "
            "RETURN labels(e) AS labels, "
            "e.name AS name, e.id AS id"
        )
        records = await self._execute(
            "read", cypher, {"path": file_path, "project_id": project_id}
        )
        for record in records:
            labels = record.get("labels", [])
            entry = {
                "name": record.get("name", ""),
                "id": record.get("id", ""),
            }
            if "Class" in labels:
                result["classes"].append(entry)
            elif "Function" in labels:
                result["functions"].append(entry)
            elif "Predicate" in labels:
                result["predicates"].append(entry)
        return result

    @traces(
        spec="008",
        story="FR-010",
        reqs=["REQ-008-026", "REQ-008-082"],
    )
    async def delete_file_cascade(self, file_path: str, project_id: str) -> int:
        """Delete a File node and ALL child entities.

        Removes the File node plus all Class, Function,
        Predicate, and Rule nodes connected via DEFINES,
        along with all their relationships.

        Args:
            file_path: Path of the File node to delete.
            project_id: The owning project's id — File.path is
                project-relative and collides across projects (2026-07-12),
                so the match must be scoped to avoid cascading a delete
                into another project's entities.

        Returns:
            Count of deleted nodes.
        """
        cypher = (
            "MATCH (f:File {path: $path, project_id: $project_id}) "
            "OPTIONAL MATCH (f)-[:DEFINES]->(c:Class) "
            "OPTIONAL MATCH (f)-[:DEFINES]->(fn:Function) "
            "OPTIONAL MATCH (f)-[:DEFINES]->(p:Predicate) "
            "OPTIONAL MATCH (f)-[:DEFINES]->(r:Rule) "
            "WITH f, "
            "collect(DISTINCT c) + "
            "collect(DISTINCT fn) + "
            "collect(DISTINCT p) + "
            "collect(DISTINCT r) AS children "
            "UNWIND children + [f] AS node "
            "WITH collect(DISTINCT node) AS nodes "
            "UNWIND nodes AS n "
            "DETACH DELETE n "
            "RETURN count(*) AS deleted"
        )
        records = await self._execute(
            "write", cypher, {"path": file_path, "project_id": project_id}
        )
        if records:
            return records[0].get("deleted", 0)
        return 0

    @traces(spec="008", story="FR-010", reqs=["REQ-008-090"])
    async def update_file_line_count(
        self, file_path: str, line_count: int, project_id: str
    ) -> bool:
        """Update the line_count property on a File node.

        Args:
            file_path: The file path to update.
            line_count: The new line count value.
            project_id: The owning project's id — File.path is
                project-relative and collides across projects (2026-07-12),
                so the match must be scoped to avoid updating another
                project's File node.

        Returns:
            True if the node was found and updated.
        """
        cypher = (
            "MATCH (f:File {path: $path, project_id: $project_id}) "
            "SET f.line_count = $line_count "
            "RETURN f.id AS id"
        )
        params = {
            "path": file_path,
            "line_count": line_count,
            "project_id": project_id,
        }
        records = await self._execute("write", cypher, params)
        return len(records) > 0