Currently the lifespan task runs in a sibling task to any request tasks, but a number of use cases require wrapping a context manager around all the request tasks, eg:
support app factories like this:
@contextlib.asynccontextmanager
async def app_factory() -> AsyncGenerator[App, None]:
async with anyio.create_task_group() as tg:
app = FastAPI(__name__)
app.include_router(items.router)
yield app
or :
var: ContextVar[int] = ContextVar('var', default=42)
@contextlib.asynccontextmanager
async def app_factory():
token = var.set(999)
try:
app = FastAPI(__name__)
app.include_router(items.router)
yield app
finally:
var.reset(token)
or including the __aenter__/__aexit__
directly on the app instance:
class ACMGRFastAPI(FastAPI):
async def __aenter__(self) -> "ACMGRFastAPI":
async with AsyncExitStack() as stack:
self.database = stack.enter_async_context(database())
self.cache = stack.enter_async_context(cache())
self.bot = stack.enter_async_context(bot())
self._stack = stack.pop_all()
return self
async def __aexit__(self, *exc_info) -> bool | None:
return await self._stack.__aexit__(*exc_info)
run lifespan as a parent task to all the request tasks
see https://gitter.im/encode/community?at=610989bc8fc359158c4f959e
Pay now to fund the work behind this issue.
Get updates on progress being made.
Maintainer is rewarded once the issue is completed.
You're funding impactful open source efforts
You want to contribute to this effort
You want to get funding like this too