INNER CODE UNIT · Python

ingest_events

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

    async def ingest_events(request: Request, payload: Any = Body(...)) -> IngestResponse:
        service: AuditQueueService = request.app.state.audit_queue
        current_settings: Settings = request.app.state.audit_settings
        events = _normalize_payload(payload)
        events = filter_records(events, current_settings.verbosity)
        if not events:
            return IngestResponse(accepted=0, queued=service.queue.qsize())
        if not service.enqueue_many(events):
            raise HTTPException(status_code=503, detail="audit queue full")
        return IngestResponse(accepted=len(events), queued=service.queue.qsize())

    @app.get("/v1/audit/events", response_model=QueryResponse)
    async def read_events(
        request: Request,
        since: str | None = Query(default=None),
        from_: str | None = Query(default=None, alias="from"),
        until: str | None = Query(default=None),
        to: str | None = Query(default=None),

View source record →