INNER CODE UNIT · Python

_log_query_failure

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

def _log_query_failure(task: asyncio.Task[QueryResult]) -> None:
    if task.cancelled():
        return
    error = task.exception()
    if error is not None and not isinstance(error, AuditQueryCancelled):
        _logger.warning("audit history scan failed", exc_info=error)


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:

View source record →