Lifespan Events and Background Tasks in FastAPI

Two common needs in API servers: running code once when the app starts or stops, and doing a bit of work after the response is sent.

Lifespan: startup and shutdown

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    app.state.http = httpx.AsyncClient(timeout=10)
    app.state.model = load_ml_model()
    yield
    # shutdown
    await app.state.http.aclose()
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

This replaces the older @app.on_event("startup") handlers. Shared clients created here (HTTP clients, connection pools) are reused across requests instead of being created each time.

from fastapi import Request

@app.get("/weather")
async def weather(city: str, request: Request):
    res = await request.app.state.http.get("https://api.example.com/weather", params={"q": city})
    return res.json()

BackgroundTasks: work after the response

from fastapi import BackgroundTasks

def send_welcome_email(email: str):
    mailer.send(to=email, subject="Welcome!", body="Thanks for signing up.")

@app.post("/signup", status_code=201)
async def signup(data: SignupRequest, background: BackgroundTasks, db: DbSession):
    user = await users.create(db, data)
    background.add_task(send_welcome_email, user.email)  # runs after the response is sent
    return {"id": user.id}

When you need a real queue

Background tasks run in the same process. If the server restarts, the task is lost, and there are no retries. Use a job queue (Celery, RQ, Dramatiq, arq) when:

• The task takes more than a few seconds.
• It must not be lost (payments, important emails).
• It needs retries, scheduling or its own scaling.