orchestrata_ultimate.semantic_graph.repository

Reference › orchestrata_ultimate › semantic_graph › repository

The factory entry point for semantic graph backends. create_semantic_graph takes an optional SemanticGraphConfig (read from the environment with from_env when omitted), looks the configured backend name up in a registry holding memory (InMemorySemanticGraph) and memgraph (MemgraphSemanticGraph), and returns the adapter, so callers never import Bolt drivers or adapter classes. The registry is the private _BACKENDS mapping. The production backend targets the shared memgraph database. The package facade and the enrichment script import it. See config and interfaces.

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/semantic_graph/repository.py

"""Repository factory for semantic graph backends.

Provides a single entry point for creating a ``SemanticGraph`` adapter based on
configuration. The factory hides backend-specific instantiation so callers do
not import Bolt drivers or adapter classes directly.
"""

from __future__ import annotations

from collections.abc import Callable

from orchestrata_ultimate.semantic_graph.adapters.memgraph import (
    MemgraphSemanticGraph,
)
from orchestrata_ultimate.semantic_graph.adapters.memory import (
    InMemorySemanticGraph,
)
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.semantic_graph.exceptions import SemanticGraphError
from orchestrata_ultimate.semantic_graph.interfaces import SemanticGraph

# Concrete constructors rather than `type[SemanticGraph]`: SemanticGraph is a
# Protocol, and `type[Protocol]` has no callable `__init__(self, config)`
# signature as far as the type checker is concerned.
_BACKENDS: dict[str, Callable[[SemanticGraphConfig], SemanticGraph]] = {
    "memory": InMemorySemanticGraph,
    "memgraph": MemgraphSemanticGraph,
}


def create_semantic_graph(
    config: SemanticGraphConfig | None = None,
) -> SemanticGraph:
    """Create a configured ``SemanticGraph`` adapter.

    Args:
        config: Backend configuration. Defaults to ``SemanticGraphConfig.from_env()``.

    Returns:
        A concrete SemanticGraph implementation matching ``config.backend``.

    Raises:
        SemanticGraphError: If the requested backend is unknown.
    """
    if config is None:
        config = SemanticGraphConfig.from_env()
    backend_cls = _BACKENDS.get(config.backend)
    if backend_cls is None:
        raise SemanticGraphError(
            f"Unknown semantic graph backend: {config.backend!r}. "
            f"Supported backends: {', '.join(sorted(_BACKENDS))}"
        )
    return backend_cls(config)