orchestrata_ultimate.core.graph.interfaces

Reference › orchestrata_ultimate › core › graph › interfaces

The protocols for the operational graph, the single store used across the gateway, kernel, session and knowledge-graph subsystems for sessions, audit trails, code intelligence, CXP packets and more. They are Protocol classes so that adapters can be swapped without inheritance, mirroring the semantic-graph seam; callers such as the semantic context provider and the knowledge-graph client depend on these and not on the driver. The design is in plans/neo4j-seam-design.md.

SyncGraphRepository has execute_read and execute_write (managed, retry-safe transactions), execute_autocommit (one statement as a true auto-commit, needed for DDL portability because Memgraph rejects index and constraint changes inside a managed transaction), unit_of_work (a context manager yielding a UnitOfWork), verify_connectivity and close. A UnitOfWork has one method, run(cypher, params); its statements share one transaction that commits on clean exit and rolls back on an exception. AsyncGraphRepository offers the async read, write, connectivity and close operations. Each execute method returns one dictionary per record and takes an optional database override. The protocols are runtime_checkable. Implementations are in graph.adapters.

At a glance

Class diagram

Dependency diagram

API

Classes

UnitOfWork (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 22

A single graph transaction spanning N statements.

SyncGraphRepository (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 46

Synchronous repository seam for the operational graph.

AsyncGraphRepository (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 149

Asynchronous repository seam for the operational graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/graph/interfaces.py

"""Repository interfaces for the operational graph.

The interfaces are Protocols so adapters can be swapped without inheritance,
mirroring ``semantic_graph.interfaces``. Unlike the semantic-graph seam (RAG /
GraphRAG, backend-agnostic Memgraph-or-memory), this seam covers the single
*operational* Neo4j store used across gateway/kernel/session/knowledge_graph
subsystems: sessions, audit trails, code intelligence, CXP packets, and more.

See ``plans/neo4j-seam-design.md`` for the full design rationale.
"""

from __future__ import annotations

from collections.abc import Mapping
from contextlib import AbstractContextManager
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces


@runtime_checkable
class UnitOfWork(Protocol):
    """A single graph transaction spanning N statements.

    Obtained from ``SyncGraphRepository.unit_of_work()``; every ``run()``
    call executes inside the same transaction, which commits on clean exit
    from the context manager and rolls back on exception.
    """

    def run(
        self, cypher: str, params: Mapping[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        """Execute one statement in the current transaction.

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

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable


@runtime_checkable
class SyncGraphRepository(Protocol):
    """Synchronous repository seam for the operational graph."""

    def execute_read(
        self,
        cypher: str,
        params: Mapping[str, Any] | None = None,
        *,
        database: str | None = None,
    ) -> list[dict[str, Any]]:
        """Run a read query in a managed (retry-safe) read transaction.

        Args:
            cypher: The Cypher query string.
            params: Query parameters.
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable

    def execute_write(
        self,
        cypher: str,
        params: Mapping[str, Any] | None = None,
        *,
        database: str | None = None,
    ) -> list[dict[str, Any]]:
        """Run a write query in a managed (retry-safe) write transaction.

        Args:
            cypher: The Cypher query string.
            params: Query parameters.
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable

    def execute_autocommit(
        self,
        cypher: str,
        params: Mapping[str, Any] | None = None,
        *,
        database: str | None = None,
    ) -> list[dict[str, Any]]:
        """Run one statement as a TRUE auto-commit transaction (no managed
        transaction function wrapping it).

        For DDL (constraint/index creation) specifically: Memgraph rejects
        constraint manipulation inside an explicit/managed transaction
        ("Constraint manipulation is not allowed in multicommand
        transactions") — Neo4j allows both. ``execute_write`` always uses a
        managed transaction (``session.execute_write``), which is correct
        for regular writes but NOT portable for DDL across both engines.
        This method is the portable DDL-application seam (Neo4j ->
        orchgraph migration, plans/neo4j-to-orchgraph-migration-plan.md
        Phase 1 Step 1.2 / ``knowledge_graph.canonical_schema.apply``).

        Args:
            cypher: The Cypher/DDL statement.
            params: Query parameters.
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable

    def unit_of_work(
        self, *, database: str | None = None
    ) -> AbstractContextManager[UnitOfWork]:
        """Open one transaction spanning multiple statements.

        Args:
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            A context manager yielding a ``UnitOfWork``. Commits on clean
            exit, rolls back on exception.
        """
        ...  # pragma: no cover - protocol method body is not executable

    def verify_connectivity(self) -> None:
        """Raise if the backend is unreachable."""
        ...  # pragma: no cover - protocol method body is not executable

    def close(self) -> None:
        """Close the underlying connection.

        A no-op when this repository wraps a driver it does not own
        (``shared`` semantics) — see the adapter docstrings.
        """
        ...  # pragma: no cover - protocol method body is not executable


@runtime_checkable
class AsyncGraphRepository(Protocol):
    """Asynchronous repository seam for the operational graph."""

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def execute_read(
        self,
        cypher: str,
        params: Mapping[str, Any] | None = None,
        *,
        database: str | None = None,
    ) -> list[dict[str, Any]]:
        """Run a read query in a managed (retry-safe) read transaction.

        Args:
            cypher: The Cypher query string.
            params: Query parameters.
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def execute_write(
        self,
        cypher: str,
        params: Mapping[str, Any] | None = None,
        *,
        database: str | None = None,
    ) -> list[dict[str, Any]]:
        """Run a write query in a managed (retry-safe) write transaction.

        Args:
            cypher: The Cypher query string.
            params: Query parameters.
            database: Target database name, or None to use the repository's
                configured/default database.

        Returns:
            One dict per result record.
        """
        ...  # pragma: no cover - protocol method body is not executable

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def verify_connectivity(self) -> None:
        """Raise if the backend is unreachable."""
        ...  # pragma: no cover - protocol method body is not executable

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def close(self) -> None:
        """Close the underlying connection.

        A no-op when this repository wraps a driver it does not own
        (``shared`` semantics) — see the adapter docstrings.
        """
        ...  # pragma: no cover - protocol method body is not executable