Background tasks and lifespan events
Run work after the response with BackgroundTasks, own your resources with the lifespan context manager, and know when a real queue is required.
BackgroundTasks
from fastapi import BackgroundTasks
def write_audit(record_id: int, actor: str) -> None:
with SessionLocal() as db: # new session: the request's is closed
db.add(AuditLog(record_id=record_id, actor=actor))
db.commit()
@app.post("/items", status_code=201)
def create(item: ItemIn, tasks: BackgroundTasks, db: Session = Depends(get_db)):
obj = Item(**item.model_dump())
db.add(obj)
db.commit()
db.refresh(obj)
tasks.add_task(write_audit, obj.id, "api")
return obj- The task runs after the response is sent, in the same process, on the same event loop (or the threadpool for a sync function).
- Never reuse the request's database session inside a background task: the dependency may already have closed it.
- A task is not persisted anywhere. If the process restarts, queued work is lost silently.
Lifespan: owning shared resources
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.pool = await asyncpg.create_pool(DSN, min_size=1, max_size=10)
app.state.model = load_model("models/ranker.bin") # expensive, once
try:
yield
finally:
await app.state.pool.close()
app = FastAPI(lifespan=lifespan)
@app.get("/score")
async def score(text: str, request: Request):
return request.app.state.model.predict(text)Code before yield runs once at startup, code after it runs at shutdown. Anything expensive and shared — connection pools, loaded models, caches — belongs here rather than in a module-level global.
When BackgroundTasks is not enough
| Need | BackgroundTasks | Task queue (Celery, RQ, Arq) |
|---|---|---|
| Send a short email after 201 | Good fit | Overkill |
| Survive a process restart | No | Yes, with a broker |
| Retry with backoff | Manual, in memory | Built in |
| Run on a separate worker pool | No | Yes |
| Scheduled at a future time | No | Yes |
| Report progress to the client | No | Via a status endpoint or WebSocket |
⚠️
BackgroundTasks keeps running work inside your web process, so CPU-heavy tasks compete with request handling and every worker scales together. With more than one uvicorn worker, the task runs in whichever worker received the request — never build state that assumes a single process.FAQ
Can I access the request object inside a background task?
Pass the values you need as arguments instead. Holding the request after the response has been sent risks using a closed body, a released session, or a client that has already disconnected.
Where should a scheduled job live?
In a separate process or a scheduler (a cron container, APScheduler in its own service, or the queue's beat feature). Starting timers in each web worker means the job runs once per worker.
Related
Databases with SQLAlchemy and session dependencies Deployment: uvicorn workers, Docker and reverse proxies
Last refreshed 2026-09-18.