orchestrata_ultimate.knowledge_graph.nodes

Reference › orchestrata_ultimate › knowledge_graph › nodes

Conversion functions between the node dataclasses of types_nodes and the label-plus-properties form that Cypher needs for the operational graph (orchkg on orchgraph). node_to_cypher_props flattens a generic Node. file_node_to_cypher, class_node_to_cypher, function_node_to_cypher, predicate_node_to_cypher and rule_node_to_cypher each return a (label, properties) tuple for their node type. cypher_record_to_node goes the other way, turning a query result record back into a Node, or None when it cannot. The functions are pure; they run no queries. graph_client and graph_node_mixins import them to write and read nodes.

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/nodes.py

"""Graph node wrappers for the knowledge graph.

Converts between Python data models and Neo4j node representations.
Includes PredicateNode and RuleNode for Prolog entities (REQ-008-018).
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.schema import (
    LABEL_CLASS,
    LABEL_FILE,
    LABEL_FUNCTION,
    LABEL_PREDICATE,
    LABEL_RULE,
)
from orchestrata_ultimate.knowledge_graph.types import (
    ClassNode,
    FileNode,
    FunctionNode,
    Node,
    NodeType,
    PredicateNode,
    RuleNode,
)

logger = logging.getLogger(__name__)


def node_to_cypher_props(node: Node) -> dict[str, Any]:
    """Convert a generic Node to Cypher-compatible properties."""
    props = {
        "id": node.id,
        "name": node.name,
        "type": node.type.value,
        "created_at": node.created_at.isoformat(),
        "updated_at": node.updated_at.isoformat(),
    }
    if node.valid_until:
        props["valid_until"] = node.valid_until.isoformat()
    props.update(node.properties)
    return props


@traces(spec="008", story="US-001", reqs=["REQ-008-021"])
def file_node_to_cypher(file_node: FileNode) -> tuple[str, dict[str, Any]]:
    """Convert FileNode to (label, properties) for Cypher."""
    props = {
        "id": file_node.id,
        "path": file_node.path,
        "name": file_node.name,
        "language": file_node.language,
        "line_count": file_node.line_count,
        "size_bytes": file_node.size_bytes,
        "hash": file_node.hash,
        "created_at": file_node.created_at.isoformat(),
        "updated_at": file_node.updated_at.isoformat(),
    }
    if file_node.project_id:
        props["project_id"] = file_node.project_id
    return LABEL_FILE, props


@traces(spec="008", story="US-001", reqs=["REQ-008-021"])
def class_node_to_cypher(class_node: ClassNode) -> tuple[str, dict[str, Any]]:
    """Convert ClassNode to (label, properties) for Cypher."""
    props = {
        "id": class_node.id,
        "name": class_node.name,
        "file_id": class_node.file_id,
        "line_start": class_node.line_start,
        "line_end": class_node.line_end,
        "docstring": class_node.docstring,
        "methods": class_node.methods,
        "base_classes": class_node.base_classes,
        "decorators": class_node.decorators,
        "created_at": class_node.created_at.isoformat(),
    }
    return LABEL_CLASS, props


@traces(spec="008", story="US-001", reqs=["REQ-008-021"])
@traces(spec="015", story="US-001", reqs=["REQ-015-120"])
def function_node_to_cypher(func_node: FunctionNode) -> tuple[str, dict[str, Any]]:
    """Convert FunctionNode to (label, properties) for Cypher."""
    props = {
        "id": func_node.id,
        "name": func_node.name,
        "file_id": func_node.file_id,
        "line_start": func_node.line_start,
        "line_end": func_node.line_end,
        "signature": func_node.signature,
        "parameters": func_node.parameters,
        "return_type": func_node.return_type,
        "decorators": func_node.decorators,
        "calls": func_node.calls,
        "is_async": func_node.is_async,
        "docstring": func_node.docstring,
        "created_at": func_node.created_at.isoformat(),
    }
    if func_node.class_id:
        props["class_id"] = func_node.class_id
    return LABEL_FUNCTION, props


@traces(spec="015", story="US-001", reqs=["REQ-015-121"])
def predicate_node_to_cypher(pred_node: PredicateNode) -> tuple[str, dict[str, Any]]:
    """Convert PredicateNode to (label, properties) for Cypher."""
    props = {
        "id": pred_node.id,
        "name": pred_node.name,
        "arity": pred_node.arity,
        "functor": pred_node.functor,
        "file_id": pred_node.file_id,
        "module": pred_node.module,
        "line_start": pred_node.line_start,
        "line_end": pred_node.line_end,
        "is_dynamic": pred_node.is_dynamic,
        "is_exported": pred_node.is_exported,
        "docstring": pred_node.docstring,
        "created_at": pred_node.created_at.isoformat(),
    }
    return LABEL_PREDICATE, props


@traces(spec="015", story="US-001", reqs=["REQ-015-122"])
def rule_node_to_cypher(rule_node: RuleNode) -> tuple[str, dict[str, Any]]:
    """Convert RuleNode to (label, properties) for Cypher."""
    props = {
        "id": rule_node.id,
        "head": rule_node.head,
        "body": rule_node.body,
        "predicate_id": rule_node.predicate_id,
        "file_id": rule_node.file_id,
        "line_start": rule_node.line_start,
        "line_end": rule_node.line_end,
        "created_at": rule_node.created_at.isoformat(),
    }
    return LABEL_RULE, props


@traces(spec="015", story="US-001", reqs=["REQ-015-123"])
def cypher_record_to_node(record: dict[str, Any]) -> Node | None:
    """Convert a Cypher query result record back to a Node."""
    if not record:
        return None
    node_type = record.get("type", "file")
    try:
        nt = NodeType(node_type)
    except ValueError as exc:
        logger.warning("Unknown node type, defaulting to FILE: %s", exc)
        nt = NodeType.FILE
    return Node(
        id=record.get("id", ""),
        type=nt,
        name=record.get("name", ""),
        properties={
            k: v
            for k, v in record.items()
            if k
            not in ("id", "type", "name", "created_at", "updated_at", "valid_until")
        },
    )