orchestrata_ultimate.postman.cli

Reference › orchestrata_ultimate › postman › cli

The command line for running the broker as a standalone process. A click group cli carries start, status (a placeholder) and version, and main is the console-script entry. No other module imports it. start takes a config path and a log level, loads the YAML configuration with _load_config (the path option or ORCHESTRATA_CONFIG; an empty configuration for a missing path or a CHANGEME_ template value; a click error if the file does not exist; SPEC-002 REQ-002-US-009-001) and runs _run_postman. That function installs SIGTERM and SIGINT handlers, builds a Postman from the core.postman section of the configuration, initialises and starts it, logs the process id, waits for the stop signal and stops it. The same broker is started inside the monolithic deployment by cli_main; this module is the stand-alone counterpart of the kernel's cli.

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

# mypy: ignore-errors
"""Orchestrata Postman CLI — standalone message router process.

Commands:
  start   Boot the postman component
  status  Show postman status

Author: Mohamed
"""

# Spec: 002 - Postman Message Broker

import asyncio
import logging
import os
import signal
from pathlib import Path
from typing import Any

import click
import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman import Postman

logger = logging.getLogger("orchestrata_ultimate.postman")

__version__ = "0.1.0"


@traces(spec="002", story="US-009", reqs=["REQ-002-US-009-001"])
def _load_config(
    config_path: str | None, environ: dict[str, str] | None = None
) -> dict[str, Any]:
    """Load YAML config.

    Args:
        config_path: Explicit config file path.
        environ: Optional environment mapping (defaults to os.environ).
    """
    env = os.environ if environ is None else environ
    path = config_path or env.get("ORCHESTRATA_CONFIG")
    # An unsubstituted .env template sentinel (CHANGEME_*) means "not
    # configured" — same convention as kernel.cli_main._load_config and
    # scripts/code_intel.py's ORCHESTRATA_ROOT guard.
    if not path or path.startswith("CHANGEME_"):
        return {}
    p = Path(path)
    if not p.exists():
        raise click.ClickException(f"Config file not found: {path}")
    with open(p) as f:
        return yaml.safe_load(f) or {}


async def _run_postman(config: dict[str, Any]) -> None:
    """Run the postman as a standalone process."""
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop_event.set)

    postman_config = config.get("core", {}).get("postman", {})
    postman = Postman(config=postman_config)
    await postman.initialize(config)
    await postman.start()

    logger.info("Postman started — PID %d", os.getpid())
    await stop_event.wait()

    logger.info("Postman shutting down...")
    await postman.stop()
    logger.info("Postman stopped")


@click.group()
def cli() -> None:
    """Orchestrata Postman — Message routing and delivery engine."""


@cli.command()
@click.option(
    "--config", "-c", "config_path", type=click.Path(), help="YAML config file"
)
@click.option(
    "--log-level",
    default="INFO",
    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"], case_sensitive=False),
)
def start(config_path: str | None, log_level: str) -> None:
    """Start the postman as a standalone process."""
    logging.basicConfig(
        level=getattr(logging, log_level.upper(), logging.INFO),
        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
    )
    config = _load_config(config_path)
    asyncio.run(_run_postman(config))


@cli.command()
def status() -> None:
    """Show postman status (placeholder)."""
    click.echo(
        "postman: not connected (use 'orchestrata health' for running instances)"
    )


@cli.command()
def version() -> None:
    """Show postman version."""
    click.echo(f"orchestrata_ultimate-postman {__version__}")


def main() -> None:
    """Entry point for console_scripts."""
    cli()


if __name__ == "__main__":
    main()