FastAPI / Chapter 6 of 8

Lifespan, background tasks, and WebSockets

Manage FastAPI startup and shutdown with lifespan, schedule background tasks, and serve WebSockets or event streams. Includes OpenAPI examples.

5 topicsUpdated 22 Oct 2026

20 / 28

BackgroundTasks and external job queues#

BackgroundTasks runs after the response in the same process. It suits small noncritical work. Durable, slow, or CPU-heavy jobs belong in a separate queue with retries.

PythonSmall post-response action

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_audit_line(message: str) -> None:
    with open("audit.log", "a", encoding="utf-8") as f:
        f.write(message + "\n")

@app.post("/jobs", status_code=202)
def submit_job(tasks: BackgroundTasks):
    job_id = "j-123"
    tasks.add_task(write_audit_line, f"accepted {job_id}")
    return {"id": job_id, "status": "accepted"}

HTTPDurable workflow idea

POST /exports → store job row → enqueue job ID → 202
worker → fetch job → generate file → update row
GET /exports/{id} → status and result URL

# Make enqueue idempotent, persist state, retry safely,
# and monitor failed jobs. A BackgroundTask is not durable.

Resource lifetime

Do not pass a request-scoped database Session into a background task. Create a fresh session inside the task or worker.

21 / 28

Lifespan: startup and shutdown resources#

Use one lifespan context for resources your process owns, such as clients and caches. It runs once per process, not once per request. Each worker process runs its own lifespan.

PythonShared HTTP client and shutdown

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with httpx.AsyncClient(timeout=5) as client:
        app.state.http = client
        yield
    # http client closes on exit

app = FastAPI(lifespan=lifespan)

@app.get("/external")
async def external(request: Request):
    r = await request.app.state.http.get("https://example.org/data")
    r.raise_for_status()
    return r.json()

PythonReadiness and liveness

from fastapi import FastAPI, HTTPException
from sqlalchemy import text
from db import Db

app = FastAPI()

@app.get("/live", include_in_schema=False)
def live():
    return {"ok": True}   # process answers

@app.get("/ready", include_in_schema=False)
def ready(db: Db):
    try:
        db.execute(text("SELECT 1"))
    except Exception:
        raise HTTPException(503, "Database unavailable")
    return {"ok": True}  # dependencies available

Shutdown

Let your container or process manager give workers time to finish or cancel requests gracefully. Keep liveness fast; a failure in a shared dependency should normally affect readiness, not force endless restarts.

22 / 28

WebSockets and streaming events#

WebSockets keep a two-way connection open. For one-way live updates, an HTTP stream may be simpler. Both require disconnect handling, authentication, and limits.

PythonMinimal WebSocket echo

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_echo(socket: WebSocket):
    # Check token/cookie and origin according to your threat model.
    await socket.accept()
    try:
        while True:
            message = await socket.receive_text()
            await socket.send_json({"echo": message})
    except WebSocketDisconnect:
        pass  # client disconnected

PythonStreaming response with disconnect handling

import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/events")
async def events(request: Request):
    async def messages():
        for number in range(3):
            if await request.is_disconnected():
                break
            yield f"data: {number}\n\n"
            await asyncio.sleep(1)
    return StreamingResponse(messages(),
                             media_type="text/event-stream")

Scaling beyond one process

An in-memory connection list reaches only one worker. Use a broker/shared pub-sub for broadcasts across workers; provide backpressure and heartbeats for long-lived connections.

23 / 28

OpenAPI, docs, responses, and API versioning#

FastAPI builds an OpenAPI contract from your route annotations and models. Add accurate examples, descriptions, response codes, and tags so generated docs and client SDKs match real behavior.

PythonDocument a route contract

from typing import Annotated
from fastapi import APIRouter, Query
from pydantic import BaseModel, Field

router = APIRouter(prefix="/v1/items", tags=["items"])

class ItemOut(BaseModel):
    id: int = Field(examples=[17])
    name: str = Field(examples=["Desk lamp"])

@router.get(
    "/{item_id}", response_model=ItemOut,
    summary="Get an item", operation_id="get_item_v1",
    responses={404: {"description": "Item not found"}},
)
def get_item(item_id: int,
             locale: Annotated[str, Query(pattern="^(en|bn)$")] = "en"):
    return ItemOut(id=item_id, name="Desk lamp")

PythonApp metadata and docs URLs

from fastapi import FastAPI

app = FastAPI(
    title="Inventory API", version="1.0.0",
    description="Manage items, stock, and orders.",
    docs_url="/docs", redoc_url="/redoc",
    openapi_url="/openapi.json",
)
app.include_router(router)

# Mount independent versions as separate routers with prefixes.
# Deprecate operations with deprecated=True; keep contracts stable.

Contract accuracy

Status codes listed in responses document possible outcomes; your handler must still raise or return those outcomes. Avoid changing schemas in place for clients that rely on them.

24 / 28

Advanced route tools: request, response, mounts, templates#

Reach for raw Request only when metadata or a body format cannot be expressed through typed parameters. Mount ASGI apps for separate subapplications or static files; templates require a template engine.

PythonRequest, cookies, and redirects

from fastapi import FastAPI, Request, Response
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/where")
def where(request: Request):
    return {"url": str(request.url),
            "client_host": request.client.host if request.client else None}

@app.post("/session")
def create_session(response: Response):
    response.set_cookie("session", "opaque-token", httponly=True,
                        secure=True, samesite="lax")
    return {"ok": True}

@app.get("/old-path")
def old_path():
    return RedirectResponse("/new-path", status_code=307)

PythonStatic files and subapplications

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()
admin = FastAPI(title="Admin API")

@admin.get("/health")
def admin_health():
    return {"ok": True}

app.mount("/admin", admin)  # separate docs at /admin/docs
app.mount("/static", StaticFiles(directory="static"),
          name="static")
# FastAPI can also return HTML via Jinja2Templates.

Cookies and CSRF

If browser authentication uses cookies, implement CSRF defenses for state-changing requests. If authentication uses an Authorization header, protect tokens against script access and configure CORS carefully.