TopolearnOpening a client or loading a model per request is slow and leaks resources; per-process setup belongs in lifespan.
Some things should exist once per process, not once per request: a database engine, an httpx.AsyncClient, a Redis pool, a loaded ML model. FastAPI's tool for this is lifespan: an async context manager passed to FastAPI(lifespan=...). Code before its yield runs at startup, before the server accepts requests; code after it runs at shutdown, once requests have stopped.
Lifespan replaces the deprecated @app.on_event("startup") and @app.on_event("shutdown") handlers. Don't mix them: when you pass lifespan, on_event handlers aren't called.
Where to keep the resources? Module globals work, but lifespan state is cleaner: yield a dict, yield {"http": client}, and each request can read it as request.state.http (every request gets a shallow copy of that dict). Wrap the lookup in a dependency so endpoints stay simple and tests can override it:
def get_http(request: Request) -> httpx.AsyncClient:
return request.state.httpThings to keep in mind:
- Each worker process runs its own lifespan. Four workers mean four pools, so size them accordingly.
- If startup raises, the server doesn't start. That's good: fail fast on bad configuration or an unreachable database.
- Only the main app's lifespan runs, not those of sub-applications mounted with
app.mount. - In tests, lifespan runs only inside
with TestClient(app) as client:.
Clean up in reverse order of creation, and put the cleanup after yield in a try/finally if you open several resources, or use contextlib.AsyncExitStack.
Example
from contextlib import asynccontextmanager
from typing import Annotated
import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.testclient import TestClient
@asynccontextmanager
async def lifespan(app: FastAPI):
client = httpx.AsyncClient(timeout=5)
print("startup: client created")
yield {"http": client} # becomes request.state.http
await client.aclose()
print("shutdown: client closed")
app = FastAPI(lifespan=lifespan)
def get_http(request: Request) -> httpx.AsyncClient:
return request.state.http
@app.get("/client")
async def which_client(http: Annotated[httpx.AsyncClient, Depends(get_http)]):
return {"closed": http.is_closed}
with TestClient(app) as c: # runs startup ... shutdown
print(c.get("/client").json()) # {'closed': False}Remember
- lifespan = async context manager: before yield is startup, after is shutdown
- Yield a dict to expose resources as request.state.<name>
- Each worker process runs its own lifespan
- Use lifespan instead of the deprecated on_event handlers
Watch out for
- Creating an httpx.AsyncClient or DB engine inside each request instead of once in lifespan.