INNER CODE UNIT · Python

_query_events

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

async def _query_events(
    request: Request, store: LocalAuditStore | S3AuditStore, query: AuditQuery, timeout_seconds: float
) -> QueryResult:
    slots: BoundedSemaphore = request.app.state.audit_read_slots
    if not slots.acquire(blocking=False):
        raise HTTPException(status_code=503, detail="audit read capacity exhausted; retry later")
    cancel_event = query.cancel_event
    assert cancel_event is not None

    def scan() -> QueryResult:
        try:
            return store.query(query)
        finally:
            # Cancelling an asyncio waiter cannot stop its thread. Capacity is
            # available only after the underlying storage scan actually exits.
            slots.release()

    task = asyncio.create_task(asyncio.to_thread(scan))

View source record →