orchestrata_ultimate.core.graph.config

Reference › orchestrata_ultimate › core › graph › config

The connection settings for the operational-graph repository seam. GraphRepositoryConfig is a dataclass with four fields: uri, auth, database and driver_kwargs. A value of None means "defer", on purpose: the config does not read KG_ROUTING_URI, KG_URI or KG_DATABASE itself, because graph_factory already resolves them at call time, and a second resolver would be a second source of truth and could cache the environment at import. A non-None value goes through to the driver factory unchanged. driver_kwargs are forwarded to the factory only when the repository builds its own driver, and are ignored when a driver is injected. from_env returns a config that defers everything. The Bolt adapters and the repository factory take it.

At a glance

Class diagram

Dependency diagram

API

Classes

GraphRepositoryConfig (dataclass)

Decorators: dataclass · Line 23

Connection config for the operational graph repository.

Source

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

"""Configuration for the operational-graph repository seam.

Unlike ``semantic_graph.config.SemanticGraphConfig`` (which resolves env vars
eagerly in ``from_env()``), this config intentionally does NOT re-read
``KG_ROUTING_URI`` / ``KG_URI`` / ``KG_DATABASE`` itself.
``core.graph_factory.create_graph`` / ``create_async_graph`` already
own that resolution (at *call* time, not import time — see the
"Read env at call time" project memory). Duplicating it here would create a
second source of truth and risk import-time env caching.

``None`` on any field means "defer to the factory's env resolution"; a
non-None value is passed straight through to ``create_graph`` /
``create_async_graph`` unchanged.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass
class GraphRepositoryConfig:
    """Connection config for the operational graph repository.

    Attributes:
        uri: Bolt URI, or None to defer to ``KG_ROUTING_URI`` / ``KG_URI``
            (resolved by ``core.graph_factory`` at driver-construction time).
        auth: (username, password) tuple, or None to defer to the factory's
            unauthenticated default (no auth is negotiated by any remaining
            graph endpoint).
        database: Target database name, or None to defer to ``KG_DATABASE``
            read from the environment at CALL time (raises if unset) — see
            ``core.graph.adapters.bolt_sync.BoltSyncRepository._session_kwargs``.
            A per-call ``database=`` override always wins over both.
        driver_kwargs: Extra keyword arguments forwarded verbatim to
            ``create_graph`` / ``create_async_graph`` when the repository
            constructs its own driver (e.g. ``connection_timeout``,
            ``max_transaction_retry_time``). Ignored when a pre-built
            driver is injected.
    """

    uri: str | None = None
    auth: tuple[str, str] | None = None
    database: str | None = None
    driver_kwargs: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_env(cls) -> GraphRepositoryConfig:
        """Return a config that defers all resolution to ``core.graph_factory``.

        This is intentionally NOT ``GraphRepositoryConfig(uri=os.environ[...], ...)``:
        env resolution belongs to ``create_graph`` / ``create_async_graph``, which
        read ``KG_ROUTING_URI`` / ``KG_URI`` at the moment
        the driver is actually constructed. Returning an all-None config here
        preserves that call-time resolution instead of caching env values into
        a config object that might outlive the process's env state (e.g. in
        long-lived test sessions that mutate env between calls).
        """
        return cls()