INNER CODE UNIT · Python

healthz

caipe-io/ai-platform-engineering · ai_platform_engineering/audit_service/main.py:238

    async def healthz() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/readyz")
    async def readyz(request: Request) -> dict[str, Any]:
        service: AuditQueueService = request.app.state.audit_queue
        if service.is_stopping:
            # Fail readiness the instant shutdown starts, not only once the
            # drain finishes — Service endpoint removal is what actually stops
            # new traffic, and that only happens once the probe fails.
            raise HTTPException(status_code=503, detail="audit-service is shutting down")
        try:
            service.store.readiness_check()
        except Exception:  # noqa: BLE001
            _logger.exception("audit storage readiness check failed")
            raise HTTPException(status_code=503, detail="storage unavailable")
        status = service.status()
        if not status["running"]:

View source record →