orchestrata_ultimate.gateway.cli

Reference › orchestrata_ultimate › gateway › cli

cli is the command-line entry point of the gateway; nothing imports it, since it is started as a process. main loads .env defaults with _load_dotenv only when no explicit environment mapping is passed, configures structured logging from ORCHESTRATA_LOG_LEVEL and ORCHESTRATA_JSON_LOGS, and parses arguments with build_parser. The serve command runs run_serve, which builds a GatewayConfig from config, creates a WebSocketGateway from server, starts it, and blocks until cancelled, then stops the gateway. The version command prints the version, and any other input prints help and exits with status 1. Passing an env mapping keeps tests hermetic. Package narrative: gateway.

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/gateway/cli.py

"""Thin CLI wrapper for the WebSocket gateway."""

import argparse
import asyncio
import logging
import os
import sys
from collections.abc import Mapping

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.gateway.config import GatewayConfig
from orchestrata_ultimate.gateway.server import WebSocketGateway


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase env contract (thin delegation).

    This used to be a private ``.env``-only walk-up that never applied the
    ``env/<profile>.env`` overlay, so a gateway started from a bare shell
    resolved the prod-BASE endpoints in ``.env``. Exactly ONE canonical
    loader may exist (``scripts/check_env_contract.py`` enforces it); this
    name is kept because it is the documented entry hook.
    """
    load_env_contract()


logger = logging.getLogger(__name__)

VERSION = "0.7.0"


def build_parser() -> argparse.ArgumentParser:
    """Build the CLI argument parser.

    Returns:
        Configured ArgumentParser with serve and version
        subcommands.
    """
    parser = argparse.ArgumentParser(
        prog="orchestrata-gateway",
        description="SPEC-014 WebSocket Gateway",
    )
    sub = parser.add_subparsers(dest="command")
    sub.add_parser("serve", help="Start the gateway")
    sub.add_parser("version", help="Show version")
    return parser


@traces(spec="004", reqs=["REQ-004-SUCC-005"])
@traces(
    spec="014",
    story="US-011",
    reqs=["REQ-014-166", "REQ-014-170", "REQ-014-101"],
)
async def run_serve(env: Mapping[str, str] | None = None) -> None:
    """Start the gateway server and block.

    Args:
        env: Optional explicit environment mapping. When provided, configuration
            is resolved from this mapping instead of ``os.environ``.
    """
    config = GatewayConfig(environ=env)
    gw = WebSocketGateway(config)
    await gw.start()
    try:
        await asyncio.Future()
    except asyncio.CancelledError:
        logger.info("Gateway serve cancelled")
    finally:
        await gw.stop()


@traces(spec="004", reqs=["REQ-004-SUCC-006"])
@traces(spec="014", reqs=["REQ-014-102"])
def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> None:
    """CLI entry point.

    Args:
        argv: Command-line arguments (defaults to sys.argv[1:]).
        env: Optional explicit environment mapping. When provided, configuration
            and logging are resolved from this mapping instead of ``os.environ``.
    """
    # Only load .env defaults into the process environment when no explicit
    # environment is supplied. Tests pass a hermetic mapping and must not
    # mutate ``os.environ``.
    if env is None:
        _load_dotenv()
    # Activate the structured-logging facility (correlation_id / volk_id /
    # per-volk trace fields) at the process entry. Human-readable by default;
    # set ORCHESTRATA_JSON_LOGS=1 for one-JSON-object-per-line output.
    from orchestrata_ultimate.core.logging import configure_logging

    environ = os.environ if env is None else env
    configure_logging(
        level=environ.get("ORCHESTRATA_LOG_LEVEL", "INFO"),
        json_format=environ.get("ORCHESTRATA_JSON_LOGS", "").strip().lower()
        in ("1", "true", "yes", "on"),
    )
    parser = build_parser()
    args = parser.parse_args(argv)
    if args.command == "serve":
        asyncio.run(run_serve(env=env))
    elif args.command == "version":
        print(f"orchestrata-gateway {VERSION}")
    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()