INNER CODE UNIT · Python

audit_status

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

    async def audit_status(request: Request) -> dict[str, Any]:
        service: AuditQueueService = request.app.state.audit_queue
        current_settings: Settings = request.app.state.audit_settings
        return {
            **service.status(),
            "storage": _storage_health(service.store, current_settings),
            "local_path": current_settings.local_path if current_settings.backend == "local" else None,
            "local_gzip": current_settings.local_gzip if current_settings.backend == "local" else None,
            "local_retention_days": current_settings.local_retention_days
            if current_settings.backend == "local"
            else None,
            "s3_bucket": current_settings.s3_bucket if current_settings.backend == "s3" else None,
            "s3_prefix": current_settings.s3_prefix if current_settings.backend == "s3" else None,
        }

    @app.post("/v1/audit/events", response_model=IngestResponse, status_code=202)
    async def ingest_events(request: Request, payload: Any = Body(...)) -> IngestResponse:
        service: AuditQueueService = request.app.state.audit_queue

View source record →