INNER CODE UNIT · Python

_purge_local_retention

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

async def _purge_local_retention(store: LocalAuditStore, retention_days: int) -> None:
    try:
        deleted = await asyncio.to_thread(store.purge_expired, retention_days=retention_days)
    except Exception as exc:  # noqa: BLE001
        _logger.warning("Failed to purge expired local audit logs: %s", exc)
        return
    if deleted:
        _logger.info("Purged %s expired local audit log files", deleted)


async def _run_local_retention_cleanup(store: LocalAuditStore, retention_days: int) -> None:
    # assisted-by Codex Codex-sonnet-4-6
    while True:
        await asyncio.sleep(_LOCAL_RETENTION_CLEANUP_INTERVAL_SECONDS)
        await _purge_local_retention(store, retention_days)


def _storage_health(store: LocalAuditStore | S3AuditStore, settings: Settings) -> dict[str, Any]:

View source record →