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

NeedBackgroundTasksTask queue (Celery, RQ, Arq)
Send a short email after 201Good fitOverkill
Survive a process restartNoYes, with a broker
Retry with backoffManual, in memoryBuilt in
Run on a separate worker poolNoYes
Scheduled at a future timeNoYes
Report progress to the clientNoVia 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.

Databases with SQLAlchemy and session dependencies Deployment: uvicorn workers, Docker and reverse proxies

Last refreshed 2026-09-18.