orchestrata_ultimate.cli

Reference › orchestrata_ultimate › cli

The FastAPI application entry point of the REST gateway, which the docstring says is used by the orchestrata-api Docker container. create_app builds a FastAPI application titled "Orchestrata Ultimate" with the documentation routes /docs and /redoc, mounts the health router from postman.health (create_health_router), and adds two plain endpoints: / returns the service name, version and a running status, and /status returns the service name, version and a healthy status. An asynchronous lifespan handler logs start-up and shutdown. A module-level app is created at import time so a server can load it by name. The only test is tests/test_toplevel_cli.py. The kernel command line is separate and is started from main.

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

"""Orchestrata Ultimate API — FastAPI application entry point.

Provides REST API gateway with health checks and service status.
Used by Docker container orchestrata-api.
"""

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

from fastapi import FastAPI

from orchestrata_ultimate.postman.health import create_health_router

logger = logging.getLogger(__name__)


@asynccontextmanager
async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Application lifespan handler."""
    logger.info("Orchestrata API starting up")
    yield
    logger.info("Orchestrata API shutting down")


def create_app() -> FastAPI:
    """Create and configure the FastAPI application.

    Returns:
        Configured FastAPI app with health and status endpoints.
    """
    app = FastAPI(
        title="Orchestrata Ultimate",
        version="1.1.1",
        docs_url="/docs",
        redoc_url="/redoc",
        lifespan=_lifespan,
    )

    app.include_router(create_health_router())

    @app.get("/")
    async def root() -> dict[str, Any]:
        """Root endpoint returning service info."""
        return {
            "service": "orchestrata-ultimate",
            "version": "1.1.1",
            "status": "running",
        }

    @app.get("/status")
    async def status() -> dict[str, Any]:
        """Service status endpoint."""
        return {
            "service": "orchestrata-ultimate",
            "version": "1.1.1",
            "status": "healthy",
        }

    return app


app = create_app()