INNER CODE UNIT · Python

lifespan

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

    async def lifespan(app: FastAPI):
        app.state.audit_store = store
        app.state.audit_queue = queue_service
        app.state.audit_settings = settings
        cleanup_task: asyncio.Task[None] | None = None
        if isinstance(store, LocalAuditStore):
            await _purge_local_retention(store, settings.local_retention_days)
            cleanup_task = asyncio.create_task(
                _run_local_retention_cleanup(store, settings.local_retention_days),
                name="audit-local-retention-cleanup",
            )
        await queue_service.start()
        try:
            yield
        finally:
            if cleanup_task is not None:
                cleanup_task.cancel()
                with suppress(asyncio.CancelledError):

View source record →