Middleware, CORS, Background Tasks, Lifespan and WebSockets

Beyond individual endpoints, a production API needs behaviour that wraps every request (logging, timing, security headers), permission for browser front ends on other domains to call it (CORS), work that continues after the response is sent (background tasks), resources created at start-up and released at shutdown (lifespan), and sometimes real-time two-way communication (WebSockets). FastAPI supports all of these.

Middleware and CORS

A middleware is a function that sees every request before it reaches an endpoint and every response on its way out: @app.middleware("http") with async def that calls await call_next(request). Use it for timing, request ids, logging and headers. Browsers block JavaScript on https://myshop.com from calling https://api.myshop.com unless the API allows that origin; add CORSMiddleware with an explicit allow_origins list (avoid "*" together with credentials).

Background Tasks and Lifespan

Declare a BackgroundTasks parameter and call background_tasks.add_task(func, *args); the task runs after the response has been sent — good for sending emails or writing audit logs. For heavy or critical jobs use a real task queue (Celery, RQ, arq). A lifespan function decorated with @asynccontextmanager runs code before the app starts serving (load a model, open a connection pool) and after yield at shutdown; pass it with FastAPI(lifespan=lifespan).

WebSockets

HTTP is request–response; a WebSocket keeps one connection open so both sides can send messages at any time — chats, live dashboards, notifications, multiplayer games. @app.websocket("/ws") receives a WebSocket; call await websocket.accept(), then loop on receive_text()/send_text() (or the JSON variants) and handle WebSocketDisconnect.

Examples

Timing middleware and CORS

Python

import time
import uuid
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.testclient import TestClient

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://webnest.example", "http://localhost:5173"],
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)

@app.middleware("http")
async def add_request_info(request: Request, call_next):
    started = time.perf_counter()
    response = await call_next(request)             # run the endpoint
    response.headers["X-Request-ID"] = str(uuid.uuid4())
    response.headers["X-Process-Time-ms"] = f"{(time.perf_counter() - started) * 1000:.1f}"
    return response

@app.get("/books")
def books():
    return ["Python Basics"]

client = TestClient(app)
r = client.get("/books", headers={"Origin": "https://webnest.example"})
print(r.json(), "X-Request-ID" in r.headers, "X-Process-Time-ms" in r.headers)
print("allowed origin:", r.headers.get("access-control-allow-origin"))

preflight = client.options("/books", headers={"Origin": "https://evil.example",
                                              "Access-Control-Request-Method": "GET"})
print("preflight from unknown origin:", preflight.status_code, preflight.headers.get("access-control-allow-origin"))

Output

['Python Basics'] True True
allowed origin: https://webnest.example
preflight from unknown origin: 400 None

Background tasks and the lifespan handler

Python

from contextlib import asynccontextmanager
from fastapi import BackgroundTasks, FastAPI
from fastapi.testclient import TestClient

events = []

@asynccontextmanager
async def lifespan(app: FastAPI):
    events.append("startup: open connection pool, load settings")
    app.state.greeting = "Welcome"
    yield                                         # the application serves requests here
    events.append("shutdown: close connection pool")

app = FastAPI(lifespan=lifespan)

def send_welcome_email(email: str):
    events.append(f"email sent to {email}")      # runs after the response is returned

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    events.append("response ready")
    return {"message": f"{app.state.greeting}, {email}!"}

with TestClient(app) as client:                   # the with block triggers lifespan events
    print(client.post("/signup?email=asha@example.com").json())
for event in events:
    print(event)

Output

{'message': 'Welcome, asha@example.com!'}
startup: open connection pool, load settings
response ready
email sent to asha@example.com
shutdown: close connection pool

A WebSocket chat endpoint

Python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.testclient import TestClient

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: dict):
        for ws in self.active:
            await ws.send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/{username}")
async def chat(websocket: WebSocket, username: str):
    await manager.connect(websocket)
    try:
        while True:
            text = await websocket.receive_text()
            await manager.broadcast({"from": username, "text": text, "online": len(manager.active)})
    except WebSocketDisconnect:
        manager.disconnect(websocket)

client = TestClient(app)
with client.websocket_connect("/ws/asha") as asha:
    with client.websocket_connect("/ws/ravi") as ravi:
        asha.send_text("Hi Ravi!")
        print("asha sees:", asha.receive_json())
        print("ravi sees:", ravi.receive_json())
        ravi.send_text("Hello!")
        print("asha sees:", asha.receive_json())

Output

asha sees: {'from': 'asha', 'text': 'Hi Ravi!', 'online': 2}
ravi sees: {'from': 'asha', 'text': 'Hi Ravi!', 'online': 2}
asha sees: {'from': 'ravi', 'text': 'Hello!', 'online': 2}

Common Mistakes

Allowing every origin with allow_origins=["*"] in production, especially with credentials.
Doing slow, blocking work in middleware, which delays every request.
Using BackgroundTasks for long or must-not-fail jobs instead of a proper task queue.
Using the deprecated @app.on_event("startup") instead of a lifespan handler.
Not handling WebSocketDisconnect, leaving dead connections in the list.

Key Points to Remember

Middleware wraps every request/response: timing, logging, request ids, headers.
CORSMiddleware lists exactly which front-end origins may call the API.
BackgroundTasks run small jobs after the response is sent.
A lifespan context manager handles start-up and shutdown resources.
WebSockets give real-time two-way communication with accept/receive/send.

Practice the examples

Change an input, predict the result, then compare it with the output. Explain why the result changes.

Use your local project environment for these examples. Codelab currently runs Python and HTML/CSS/JavaScript; framework examples may need project dependencies.