orchestrata_ultimate.gateway.handlers.utils

Reference › orchestrata_ultimate › gateway › handlers › utils

utils holds one helper shared by the graph-reading handlers: _neo4j_value_to_dict converts a graph record value into a JSON-serialisable structure. It handles nodes (id, labels and properties), relationships (type and endpoints), paths, temporal values and plain dict-like objects, recursing into nested values. It is imported by the knowledge, projects and _session_kg modules so their replies can be sent as JSON. See gateway.

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/gateway/handlers/utils.py

"""Shared utilities for gateway RPC handlers."""

from __future__ import annotations

from typing import Any


def _neo4j_value_to_dict(value: Any) -> Any:
    """Convert a graph value to a JSON-serializable dict.

    Handles Neo4j Node, Relationship, Path, temporal types, and
    plain Python dict-like objects.

    Args:
        value: A Neo4j record value (Node, Relationship, etc).

    Returns:
        JSON-serializable representation.
    """
    # Neo4j Node
    if hasattr(value, "labels") and hasattr(value, "element_id"):
        result = {
            "id": value.element_id,
            "labels": list(value.labels),
        }
        for k, v in value.items():
            result[k] = _neo4j_value_to_dict(v)
        return result
    # Neo4j Relationship
    if hasattr(value, "type") and hasattr(value, "start_node"):
        result = {
            "type": value.type,
            "start": (
                value.start_node.element_id
                if hasattr(value.start_node, "element_id")
                else None
            ),
            "end": (
                value.end_node.element_id
                if hasattr(value.end_node, "element_id")
                else None
            ),
        }
        for k, v in value.items():
            result[k] = _neo4j_value_to_dict(v)
        return result
    # Neo4j Path
    if hasattr(value, "nodes") and hasattr(value, "relationships"):
        return {
            "nodes": [_neo4j_value_to_dict(n) for n in value.nodes],
            "relationships": [_neo4j_value_to_dict(r) for r in value.relationships],
        }
    # Plain dict-like objects (dicts, Neo4j Records/Entities with .items())
    if hasattr(value, "items") and callable(getattr(value, "items", None)):
        result = {}
        for k, v in value.items():
            result[k] = _neo4j_value_to_dict(v)
        return result
    # Handle Neo4j temporal types
    if hasattr(value, "iso_format"):
        return value.iso_format()
    if hasattr(value, "year") and hasattr(value, "month"):
        # DateTime, Date, Time, etc.
        return str(value)
    return value