orchestrata_ultimate.knowledge_graph.server

Reference › orchestrata_ultimate › knowledge_graph › server

The gRPC entry point for the knowledge graph service, run as the orchestrata-knowledge-graph container. main first calls load_env_contract (RUL-019), so a bare-shell launch works as well as one under systemd, then runs _serve with asyncio.run and exits quietly with status 0 on a keyboard interrupt. _serve configures logging, reads the port through require_port("KG_GRPC_PORT", env), builds a CXPServerConfig bound to all interfaces, creates the server with create_cxp_server, registers SIGTERM and SIGINT handlers that stop it, and hands over to run_serve_lifecycle from the cxp package, which guarantees the server is stopped before the coroutine returns. The optional env mapping lets tests bind an ephemeral port per test. The module issues no graph queries itself; no other module imports it, and tests/knowledge_graph/test_server.py covers it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Knowledge Graph gRPC server entry point.

Provides a gRPC-based knowledge graph service using the CXP protocol.
Used by Docker container orchestrata-knowledge-graph.
"""

import asyncio
import logging
import signal
import sys
from collections.abc import Mapping

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.cxp.endpoint_config import require_port
from orchestrata_ultimate.cxp.servicer import (
    CXPServerConfig,
    run_serve_lifecycle,
)
from orchestrata_ultimate.cxp.transport import create_cxp_server

logger = logging.getLogger(__name__)


async def _serve(env: Mapping[str, str] | None = None) -> None:
    """Start the knowledge graph gRPC server.

    Args:
        env: Optional explicit environment mapping, forwarded to
            ``require_port``. Defaults to ``os.environ``. This is the seam that
            lets a caller name the port instead of inheriting a process-wide
            one — which is how the tests bind an EPHEMERAL port per test rather
            than sharing one fixed port across every worker in the suite.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
    )

    port = require_port("KG_GRPC_PORT", env)
    config = CXPServerConfig(host="0.0.0.0", port=port)
    server = create_cxp_server(config)

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, lambda: asyncio.ensure_future(server.stop()))

    # `run_serve_lifecycle` guarantees `server.stop()` runs before this
    # coroutine returns or raises -- see its docstring for why that matters
    # (a leaked half-stopped `grpc.aio.Server` otherwise hangs the whole
    # process at interpreter exit once grpc's `shutdown_grpc_aio` trips on
    # it).
    await run_serve_lifecycle(server, label="Knowledge Graph server", port=port)


def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> None:
    """Entry point for knowledge graph server.

    Args:
        argv: Command-line arguments (unused, for compatibility).
        env: Optional explicit environment mapping, forwarded to ``_serve``.
            Defaults to ``os.environ``.
    """
    # RUL-019: load the two-phase env contract before _serve() resolves
    # KG_GRPC_PORT through require_port. This entrypoint had no loader in its
    # import closure at all, so a bare-shell launch raised on the missing key
    # and it only ever worked under systemd's EnvironmentFile= injection.
    load_env_contract()
    try:
        asyncio.run(_serve(env))
    except KeyboardInterrupt:
        logger.info("Knowledge Graph server interrupted")
        sys.exit(0)


if __name__ == "__main__":
    main()