INNER CODE UNIT · Python

readyz

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

    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"]:
            raise HTTPException(status_code=503, detail="audit worker is not running")
        return {"status": "ready", **status}

    @app.get("/v1/audit/status")

View source record →