orchestrata_ultimate.knowledge_graph.relationships

Reference › orchestrata_ultimate › knowledge_graph › relationships

Pure builders for the relationship Cypher used by the operational knowledge graph (orchkg on orchgraph). rel_type_to_neo4j maps a RelationshipType enum member from types_enums to the relationship string (the function name keeps the old Bolt-driver vocabulary). build_create_rel_cypher returns a MERGE statement and its parameters for a source label, target label, relationship type and optional properties, and build_delete_rel_cypher returns the matching delete statement. The helpers contains_rel, defines_rel, calls_rel, inherits_rel, imports_rel and depends_on_rel each produce the parameter dictionary for one edge kind from two entity ids. None of them runs a query: they only build text and dictionaries for a caller such as graph_client. No module imports it; it is exercised by the knowledge_graph schema, interface and performance tests.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Graph relationship wrappers for the knowledge graph.

Defines relationship creation helpers and type mappings.
"""

# Spec: 008 - Knowledge Graph

from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.schema import (
    REL_CALLS,
    REL_CONTAINS,
    REL_DEFINES,
    REL_DEPENDS_ON,
    REL_IMPORTS,
    REL_INHERITS,
)
from orchestrata_ultimate.knowledge_graph.types import RelationshipType


def rel_type_to_neo4j(rel_type: RelationshipType) -> str:
    """Map RelationshipType enum to the graph relationship string."""
    return rel_type.value


@traces(spec="008", story="US-001", reqs=["REQ-008-012", "REQ-008-023", "REQ-008-053"])
@traces(spec="015", story="US-001", reqs=["REQ-015-160"])
def build_create_rel_cypher(
    source_label: str,
    target_label: str,
    rel_type: str,
    properties: dict[str, Any] | None = None,
) -> tuple[str, dict[str, Any]]:
    """Build a MERGE Cypher statement for a relationship.

    Returns (cypher_query, parameters).
    """
    props_clause = ""
    params: dict[str, Any] = {}

    if properties:
        prop_parts = []
        for i, (k, v) in enumerate(properties.items()):
            param_name = f"rp_{i}"
            prop_parts.append(f"{k}: ${param_name}")
            params[param_name] = v
        props_clause = " {" + ", ".join(prop_parts) + "}"

    cypher = (
        f"MATCH (a:{source_label} {{id: $source_id}}), "
        f"(b:{target_label} {{id: $target_id}}) "
        f"MERGE (a)-[r:{rel_type}{props_clause}]->(b) "
        "RETURN type(r) AS rel_type"
    )
    return cypher, params


@traces(spec="008", story="US-001", reqs=["REQ-008-053"])
@traces(spec="015", story="US-001", reqs=["REQ-015-161"])
def build_delete_rel_cypher(
    source_label: str,
    target_label: str,
    rel_type: str,
) -> str:
    """Build a DELETE Cypher statement for a relationship."""
    return (
        f"MATCH (a:{source_label} {{id: $source_id}})"
        f"-[r:{rel_type}]->"
        f"(b:{target_label} {{id: $target_id}}) "
        "DELETE r"
    )


# Common relationship builders


def contains_rel(container_id: str, contained_id: str) -> dict[str, Any]:
    """Build CONTAINS relationship parameters."""
    return {
        "source_id": container_id,
        "target_id": contained_id,
        "rel_type": REL_CONTAINS,
    }


def defines_rel(file_id: str, entity_id: str) -> dict[str, Any]:
    """Build DEFINES relationship parameters."""
    return {"source_id": file_id, "target_id": entity_id, "rel_type": REL_DEFINES}


def calls_rel(caller_id: str, callee_id: str) -> dict[str, Any]:
    """Build CALLS relationship parameters."""
    return {"source_id": caller_id, "target_id": callee_id, "rel_type": REL_CALLS}


def inherits_rel(child_id: str, parent_id: str) -> dict[str, Any]:
    """Build INHERITS relationship parameters."""
    return {"source_id": child_id, "target_id": parent_id, "rel_type": REL_INHERITS}


def imports_rel(importer_id: str, imported_id: str) -> dict[str, Any]:
    """Build IMPORTS relationship parameters."""
    return {"source_id": importer_id, "target_id": imported_id, "rel_type": REL_IMPORTS}


def depends_on_rel(source_id: str, target_id: str) -> dict[str, Any]:
    """Build DEPENDS_ON relationship parameters."""
    return {"source_id": source_id, "target_id": target_id, "rel_type": REL_DEPENDS_ON}