INNER CODE UNIT · Python

_LOCAL_RETENTION_CLEANUP_INTERVAL_SECONDS

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

_LOCAL_RETENTION_CLEANUP_INTERVAL_SECONDS = 60 * 60
_logger = logging.getLogger(__name__)


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")

View source record →