orchestrata_ultimate.core.graph.repository

Reference › orchestrata_ultimate › core › graph › repository

The factory that creates graph repositories, the entry point callers use (the knowledge-graph conversation consumer and semantic bridge, and several scripts). create_graph_repository(config, driver=None) returns a SyncGraphRepository and create_async_graph_repository(config, driver=None) returns an AsyncGraphRepository. A config of None gives a default GraphRepositoryConfig; an injected driver is used as given and is not closed by the repository. Two private tables, _SYNC_BACKENDS and _ASYNC_BACKENDS, map the single backend key bolt to BoltSyncRepository and BoltAsyncRepository, in the same shape as the semantic-graph factory so that more backends could be added without changing callers. The key is internal: neither function takes a backend argument, and no configuration supplies one. The key was renamed from neo4j to bolt in the SPEC-023 graph rename.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

"""Repository factory for the operational-graph seam.

Provides the single entry points for creating a ``SyncGraphRepository`` /
``AsyncGraphRepository`` adapter, mirroring
``semantic_graph.repository.create_semantic_graph``. Only one backend
("bolt") exists today; the ``_SYNC_BACKENDS``/``_ASYNC_BACKENDS`` dicts
mirror semantic_graph's shape so additional backends can slot in without
changing callers.

The backend key is internal-only: neither ``create_graph_repository`` nor
``create_async_graph_repository`` exposes a ``backend=`` parameter, and no
config/env value ever supplies a key into ``_SYNC_BACKENDS``/
``_ASYNC_BACKENDS`` (unlike ``semantic_graph``'s equivalent, which IS
selected via ``backend="memgraph"``/etc. call-site kwargs) — so renaming
the key from the historical ``"neo4j"`` to the protocol-named ``"bolt"``
(SPEC-023 graph-rename, P1) is a pure rename with no external contract to
preserve.
"""

from __future__ import annotations

from collections.abc import Callable

from neo4j import AsyncDriver, Driver

from orchestrata_ultimate.core.graph.adapters.bolt_async import (
    BoltAsyncRepository,
)
from orchestrata_ultimate.core.graph.adapters.bolt_sync import (
    BoltSyncRepository,
)
from orchestrata_ultimate.core.graph.config import GraphRepositoryConfig
from orchestrata_ultimate.core.graph.interfaces import (
    AsyncGraphRepository,
    SyncGraphRepository,
)

# Concrete constructors rather than `type[SyncGraphRepository]`: the
# interfaces are Protocols, and `type[Protocol]` has no callable
# `__init__(self, config, *, driver=None)` signature as far as the type
# checker is concerned. See semantic_graph/repository.py for the precedent.
_SYNC_BACKENDS: dict[
    str, Callable[[GraphRepositoryConfig, Driver | None], SyncGraphRepository]
] = {
    "bolt": lambda config, driver: BoltSyncRepository(config, driver=driver),
}

_ASYNC_BACKENDS: dict[
    str,
    Callable[[GraphRepositoryConfig, AsyncDriver | None], AsyncGraphRepository],
] = {
    "bolt": lambda config, driver: BoltAsyncRepository(config, driver=driver),
}

_DEFAULT_BACKEND = "bolt"


def create_graph_repository(
    config: GraphRepositoryConfig | None = None,
    *,
    driver: Driver | None = None,
) -> SyncGraphRepository:
    """Create a configured ``SyncGraphRepository`` adapter.

    Args:
        config: Repository configuration. Defaults to
            ``GraphRepositoryConfig.from_env()``.
        driver: An existing ``neo4j.Driver`` to wrap (shared, not owned).
            When None, the adapter lazily creates and owns its own driver.

    Returns:
        A concrete ``SyncGraphRepository`` implementation.
    """
    if config is None:
        config = GraphRepositoryConfig.from_env()
    backend_fn = _SYNC_BACKENDS[_DEFAULT_BACKEND]
    return backend_fn(config, driver)


def create_async_graph_repository(
    config: GraphRepositoryConfig | None = None,
    *,
    driver: AsyncDriver | None = None,
) -> AsyncGraphRepository:
    """Create a configured ``AsyncGraphRepository`` adapter.

    Args:
        config: Repository configuration. Defaults to
            ``GraphRepositoryConfig.from_env()``.
        driver: An existing ``neo4j.AsyncDriver`` to wrap (shared, not
            owned). When None, the adapter lazily creates and owns its own
            driver.

    Returns:
        A concrete ``AsyncGraphRepository`` implementation.
    """
    if config is None:
        config = GraphRepositoryConfig.from_env()
    backend_fn = _ASYNC_BACKENDS[_DEFAULT_BACKEND]
    return backend_fn(config, driver)