orchestrata_ultimate.postman.health

Reference › orchestrata_ultimate › postman › health

Health checks for the broker's dependencies, SPEC-002 story FR-009 (REQ-002-FR9-001, -007). HealthChecker tests the two services the broker relies on, Redis and the operational graph, and create_health_router exposes the result as FastAPI endpoints. The postman cli side and the gateway HTTP surface (gateway.http_surface) import it. The monolithic deployment's own health server is a separate, simpler one in cli_main.

check_redis creates an async Redis client through core.redis_factory (or uses an injected one), pings it, closes it and returns {"status": "healthy"|"unhealthy", "service": "redis"}; any exception returns unhealthy with the error text. check_graph does the equivalent for the graph through the shared graph factory. get_overall_status runs both and combines them: any unhealthy check makes the whole status unhealthy, a check that could not run because its client library is not installed is reported as unknown and makes the status degraded, and otherwise it is healthy. The router has the prefix /health with /live (always alive, with a timestamp), /ready (returns the status for healthy and degraded and raises HTTP 503 for unhealthy) and /checks (lists the check names).

At a glance

Class diagram

Dependency diagram

API

Classes

HealthChecker

Decorators: traces(spec='002', story='FR-009', reqs=['REQ-002-FR9-007']) · Line 37

Simple health checker for Redis and the operational graph.

Functions

Source

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

# mypy: ignore-errors
"""Health checks for Redis/the operational graph and liveness/readiness
endpoints.

Provides health checking with FastAPI router.
"""

# Spec: 002 - Postman Message Broker

import logging
from collections.abc import Awaitable
from datetime import datetime
from typing import Any, cast

from fastapi import APIRouter, HTTPException

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_uri

logger = logging.getLogger(__name__)


def _redis_result[T](value: Awaitable[T] | T) -> Awaitable[T]:
    """Narrow a redis-py command result back to its true async type.

    redis-py types command methods (``ping``, ...) as ``Awaitable[T] | T``
    because the same command mixin backs both the sync and async clients.
    ``HealthChecker.check_redis`` only ever holds an async client (injected
    or built via ``create_async_redis``), which always returns an awaitable
    at runtime -- this documents that fact for pyright instead of
    suppressing the check (precedent: memory/hot_store.py).
    """
    return cast(Awaitable[T], value)


@traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
class HealthChecker:
    """Simple health checker for Redis and the operational graph."""

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        graph_uri: str | None = None,
        graph_username: str | None = None,
        graph_password: str | None = None,
        redis_client: Any | None = None,
    ):
        """Initialize the health checker.

        Args:
            redis_client: Pre-built async client injected directly (the
                production path still lazily builds one from
                ``redis_host``/``redis_port`` on each ``check_redis`` call
                when this is omitted). Exposed so tests can point
                ``check_redis`` at a fail-fast client without paying
                redis-py's default connection-retry backoff.
        """
        self._redis_host = redis_host
        self._redis_port = redis_port
        self._graph_uri = graph_uri
        self._graph_username = graph_username
        self._graph_password = graph_password
        self._redis_client = redis_client

    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def check_redis(self) -> dict[str, Any]:
        """Check Redis connection."""
        try:
            client = self._redis_client
            if client is None:
                from orchestrata_ultimate.core.redis_factory import (
                    create_async_redis,
                )

                kwargs: dict[str, Any] = {"decode_responses": False}
                if self._redis_host:
                    kwargs["host"] = self._redis_host
                if self._redis_port:
                    kwargs["port"] = self._redis_port
                client = create_async_redis(**kwargs)
            pong = await _redis_result(client.ping())
            await client.aclose()  # pragma: no cover
            return {
                "status": "healthy" if pong else "unhealthy",
                "service": "redis",
            }  # pragma: no cover
        except Exception as e:
            return {"status": "unhealthy", "service": "redis", "error": str(e)}

    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def check_graph(self) -> dict[str, Any]:
        """Check graph connection."""
        try:
            from orchestrata_ultimate.core.graph import (
                GraphRepositoryConfig,
                create_async_graph_repository,
            )

            repo = create_async_graph_repository(
                GraphRepositoryConfig(
                    uri=self._graph_uri or resolve_uri(),
                    auth=(
                        self._graph_username or "",
                        self._graph_password or "",
                    ),
                )
            )
            await repo.verify_connectivity()
            await repo.close()  # pragma: no cover
            return {"status": "healthy", "service": "graph"}  # pragma: no cover
        except Exception as e:
            return {"status": "unhealthy", "service": "graph", "error": str(e)}

    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def get_overall_status(self) -> dict[str, Any]:
        """Run all health checks."""
        checks = []
        # Redis check if available
        try:
            import redis.asyncio  # noqa: F401

            checks.append(await self.check_redis())
        except ImportError:
            logger.debug("Redis package not installed, skipping check")
            checks.append(
                {
                    "status": "unknown",
                    "service": "redis",
                    "error": "redis not installed",
                }
            )

        # Graph check if available
        try:
            import neo4j  # noqa: F401

            checks.append(await self.check_graph())
        except ImportError:
            logger.debug("neo4j package not installed, skipping check")
            checks.append(
                {
                    "status": "unknown",
                    "service": "graph",
                    "error": "neo4j not installed",
                }
            )

        # Determine overall status
        status = "healthy"
        for check in checks:
            if check["status"] == "unhealthy":
                status = "unhealthy"
            elif check["status"] == "unknown" and status == "healthy":
                status = "degraded"

        return {
            "status": status,
            "timestamp": datetime.now().isoformat(),
            "checks": checks,
        }


@traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-001"])
def create_health_router(checker: HealthChecker | None = None) -> APIRouter:
    """Create FastAPI router for health endpoints."""
    router = APIRouter(prefix="/health", tags=["health"])
    checker = checker or HealthChecker()

    @router.get("/live")
    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def liveness() -> dict[str, Any]:
        """Liveness probe."""
        return {"status": "alive", "timestamp": datetime.now().isoformat()}

    @router.get("/ready")
    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def readiness() -> dict[str, Any]:
        """Readiness probe."""
        status = await checker.get_overall_status()
        if status["status"] in ["healthy", "degraded"]:
            return status
        raise HTTPException(status_code=503, detail=status)

    @router.get("/checks")
    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def list_checks() -> dict[str, Any]:
        """List available health checks."""
        return {"checks": ["redis", "graph"], "timestamp": datetime.now().isoformat()}

    return router


@traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-001"])
def get_health_router() -> APIRouter:
    return create_health_router()