INNER CODE UNIT · Python

_run_local_retention_cleanup

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

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]:
    # assisted-by Codex Codex-sonnet-4-6
    try:
        if isinstance(store, LocalAuditStore):
            return store.storage_health(
                warning_percent=settings.local_disk_warning_percent,
                critical_percent=settings.local_disk_critical_percent,
            )
        return store.storage_health()
    except Exception:  # noqa: BLE001
        _logger.exception("audit storage health check failed")

View source record →