orchestrata_ultimate.core.graph_factory

Reference › orchestrata_ultimate › core › graph_factory

The single place where the operational knowledge-graph driver is built and where its connection settings are resolved. It is used widely: the graph repository seam in core.graph, its Bolt adapters, the architect volk context, and many tests all import it. It creates real Bolt-protocol drivers only; there is no in-memory or fake fallback.

create_graph and create_async_graph resolve the URI, resolve the authentication, set a driver-level transaction retry budget of MAX_TRANSACTION_RETRY_SECONDS (5 seconds, so that the project's own bounded retry in core.graph.retry is not nested inside a 30-second driver retry) and return a neo4j driver. Authentication is off unless the caller passes a tuple or an Auth; the NO_AUTH sentinel records that an endpoint is unauthenticated on purpose.

Three resolvers follow the no-literal-fallback rule of infra_env_rule (RUL-003). resolve_uri (the public form of _resolve_uri) reads KG_ROUTING_URI then KG_URI, and resolve_kg_database reads KG_DATABASE; each raises RuntimeError when the value is unset, because a silent default once pointed writes at the wrong database. resolve_kg_dialect returns memgraph when KG_ENGINE is memgraph and neo4j otherwise, so writers emit DDL the engine accepts. All three read the environment at call time.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

Truncated at 25 nodes: showing 0 of 0 imports and 24 of 45 importers.

API

Constants

Functions

Source

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

"""Graph driver factory.

Creates sync or async Bolt-protocol graph drivers from environment
configuration. Real drivers only — no in-memory or fake fallbacks.
"""

from __future__ import annotations

import os
from collections.abc import Mapping
from typing import Any

from neo4j import AsyncGraphDatabase, Auth, GraphDatabase, basic_auth

# Sentinel for connections that are EXPLICITLY unauthenticated (e.g. Memgraph
# nodes with no auth negotiation). It used to distinguish "no auth, on purpose"
# from auth=None ("unset — fall back to NEO4J_USERNAME/NEO4J_PASSWORD"). Since
# the 2026-07-27 decommission there is no authenticated graph left to fall back
# to and both resolve to None, so the sentinel now records INTENT at the call
# site rather than selecting behaviour. It is kept for exactly that: `auth=None`
# reads as "I didn't think about it", `auth=NO_AUTH` as "this endpoint has no
# auth". Call sites passing a real (user, password) tuple still authenticate.
NO_AUTH: Any = object()

#: Driver-level managed-transaction retry budget, in seconds (neo4j's own
#: default is 30). Lowered because this codebase already owns a bounded,
#: jittered MVCC retry (``core.graph.retry``), so the driver's budget is
#: belt-and-braces on the ONE failure mode it can fix, and pure latency on the
#: rest: orchgraph reports a MISSING DATABASE as a ``TransientError``, so the
#: driver's 30s and this codebase's 5 attempts nested, turning a misconfigured
#: ``database=`` into minutes of silence per call (measured live 2026-07-26 —
#: it hung tests/memory/test_cold_store_extended.py's error-path class past a
#: 180s per-test timeout). Callers may still pass their own value.
MAX_TRANSACTION_RETRY_SECONDS = 5.0


def _resolve_auth(auth: tuple[str, str] | Auth | None) -> Any:
    """Resolve graph auth; unspecified means UNAUTHENTICATED.

    This used to fall back to ``NEO4J_USERNAME``/``NEO4J_PASSWORD`` (defaulting
    to ``("neo4j", "")``). Both env keys were retired with the Neo4j containers
    on 2026-07-27 — every remaining graph endpoint is an orchgraph/Memgraph node
    that negotiates no auth, so the fallback could only ever send credentials
    nothing asks for. Verified live against both the production coordinators and
    the dedicated test node: ``auth=None`` and ``basic_auth("neo4j", "")`` are
    equally accepted, so dropping the fallback is not a behaviour change.

    An explicitly-passed tuple or ``neo4j.Auth`` is still honoured unchanged —
    this removes a default, not the ability to authenticate.

    Args:
        auth: Explicit credentials, the :data:`NO_AUTH` sentinel, or None.

    Returns:
        A ``neo4j.Auth``, or None for an unauthenticated connection.
    """
    if auth is NO_AUTH or auth is None:
        return None
    return basic_auth(*auth) if isinstance(auth, tuple) else auth


def _resolve_uri(
    uri: str | None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Resolve the graph URI from env when not provided explicitly.

    ``KG_ROUTING_URI`` / ``KG_URI`` are the canonical seam added for the
    Neo4j -> orchgraph migration (plans/neo4j-to-orchgraph-migration-plan.md
    Step 0.2). ``NEO4J_URI`` was a read-through fallback during the migration
    so unmigrated call sites kept working; Step 5.1 removed it along with the
    Neo4j containers. Keeping it would have been worse than useless — the
    value it named (``bolt://10.53.1.11:7687``) now resolves to nothing, so
    a stale export could only redirect a correctly-configured process at a
    destroyed host. Blank-but-set values (e.g. an unfilled ``.env``
    placeholder) are treated as unset.

    No literal fallback (RUL-003). ``bolt://localhost:7687`` used to be the
    last resort; nothing in this infrastructure has ever listened on
    localhost (every graph is an Incus container on 10.53.1.x), so it
    could only ever turn a config gap into a connection-refused that reads
    like an outage. Raising names the actual problem.

    Raises:
        RuntimeError: when no URI is given and none of the env keys is set.
    """
    if uri is not None:
        return uri
    env_vars = os.environ if env is None else env
    resolved = env_vars.get("KG_ROUTING_URI") or env_vars.get("KG_URI")
    if not resolved:
        raise RuntimeError(
            "No graph URI configured — set KG_ROUTING_URI or KG_URI (see "
            ".env); this code has no default address. NEO4J_URI is no longer "
            "read: it was retired with the Neo4j containers on 2026-07-27"
        )
    return resolved


def resolve_uri(
    uri: str | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Public alias for :func:`_resolve_uri` — THE graph-URI resolver.

    Exists so call sites stop hand-rolling
    ``os.environ.get("NEO4J_URI", "bolt://localhost:7687")``: each of those
    carried its own silent localhost default, and after the orchgraph
    cutover a wrong address is no longer a loud failure everywhere.
    """
    return _resolve_uri(uri, env)


def resolve_kg_database(
    database: str | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Resolve the target graph database name at call time.

    An explicit ``database`` argument always wins. Otherwise reads
    ``KG_DATABASE`` from the environment — the multi-tenancy seam added for
    the Neo4j -> orchgraph migration (plans/neo4j-to-orchgraph-migration-
    plan.md Phase 2 Step 2.1: per-worker test databases and the production
    ``orchkg`` database both flow through this single resolver). Read at
    CALL time (not cached at import/construction time) per the project's
    env-config convention: the same process may need a different answer
    across test-worker env mutations.

    No literal fallback (RUL-003), for the same reason :func:`resolve_uri`
    has none. The old ``or "neo4j"`` default was silent and wrong on this
    cluster: ``neo4j`` is not the operational KG, so an unset
    ``KG_DATABASE`` did not fail — it opened a database of that name and
    wrote there. It had already happened: on 2026-07-26 the production
    orchgraph cluster carried a ``neo4j`` database holding 56 nodes of
    memory-store TEST data (``test_755f8791_*``), invisible to every query
    that correctly targeted ``orchkg``. Raising names the config gap.

    Raises:
        RuntimeError: when no database is given and ``KG_DATABASE`` is unset.
    """
    if database is not None:
        return database
    env_vars = os.environ if env is None else env
    resolved = env_vars.get("KG_DATABASE")
    if not resolved:
        raise RuntimeError(
            "No KG database configured — set KG_DATABASE (see .env); this "
            "code has no default database name"
        )
    return resolved


def resolve_kg_dialect(
    dialect: str | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Resolve which Cypher DIALECT to emit for the operational KG.

    The two engines do not accept the same DDL — Memgraph rejects
    ``IF NOT EXISTS`` on every index/constraint form and has no
    ``CREATE FULLTEXT INDEX`` at all — so a writer that guesses wrong does
    not degrade, it raises. Before this resolver existed, ``KG_ENGINE`` was
    read ad-hoc in five places with the same inline default, and
    ``GraphSyncer`` simply defaulted to ``"neo4j"``: after the cutover its
    ``create_schema()`` failed outright on ``CREATE TEXT INDEX ... IF NOT
    EXISTS`` against an engine that had been Memgraph for hours.

    An explicit ``dialect`` argument always wins. Otherwise ``KG_ENGINE`` is
    read at CALL time (never cached at import) so per-process and per-test
    overrides apply, falling back to ``"neo4j"``.

    Returns:
        ``"memgraph"`` or ``"neo4j"``.
    """
    if dialect is not None:
        return dialect
    env_vars = os.environ if env is None else env
    return "memgraph" if env_vars.get("KG_ENGINE") == "memgraph" else "neo4j"


def create_async_graph(
    uri: str | None = None,
    auth: tuple[str, str] | Auth | None = None,
    **kwargs: Any,
) -> Any:
    """Create a real async graph driver.

    Args:
        uri: Graph URI (default: from KG_ROUTING_URI / KG_URI env).
        auth: (user, password) tuple, or a pre-built neo4j.Auth (e.g. from
            neo4j.basic_auth/kerberos_auth/bearer_auth), passed through unchanged
            (default: unauthenticated — every remaining endpoint is orchgraph).
        **kwargs: Additional driver kwargs forwarded to neo4j.AsyncGraphDatabase.driver.

    Returns:
        neo4j.AsyncDriver instance.
    """
    resolved_uri = _resolve_uri(uri)
    resolved_auth = _resolve_auth(auth)
    kwargs.setdefault("max_transaction_retry_time", MAX_TRANSACTION_RETRY_SECONDS)
    return AsyncGraphDatabase.driver(resolved_uri, auth=resolved_auth, **kwargs)


def create_graph(
    uri: str | None = None,
    auth: tuple[str, str] | Auth | None = None,
    **kwargs: Any,
) -> Any:
    """Create a real sync graph driver.

    Args:
        uri: Graph URI (default: from KG_ROUTING_URI / KG_URI env).
        auth: (user, password) tuple, or a pre-built neo4j.Auth (e.g. from
            neo4j.basic_auth/kerberos_auth/bearer_auth), passed through unchanged
            (default: unauthenticated — every remaining endpoint is orchgraph).
        **kwargs: Additional driver kwargs forwarded to neo4j.GraphDatabase.driver.

    Returns:
        neo4j.Driver instance.
    """
    resolved_uri = _resolve_uri(uri)
    resolved_auth = _resolve_auth(auth)
    kwargs.setdefault("max_transaction_retry_time", MAX_TRANSACTION_RETRY_SECONDS)
    return GraphDatabase.driver(resolved_uri, auth=resolved_auth, **kwargs)