Background & WebSockets — Cheat Sheet

BackgroundTasks — Work After the Response

import logging, time
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()
log = logging.getLogger("tasks")

def send_welcome_email(email: str, name: str):
    time.sleep(2)                       # SMTP being SMTP (sync is fine here —
    log.info("welcome mail sent to %s", email)   # it runs in the thread pool)

def audit(event: str, **fields):
    log.info("AUDIT %s %s", event, fields)

@app.post("/register", status_code=201)
def register(payload: dict, background: BackgroundTasks):
    user = {"id": 101, "email": payload["email"], "name": payload["name"]}
    # ... hash password, insert user (the REQUEST work) ...

    background.add_task(send_welcome_email, user["email"], user["name"])
    background.add_task(audit, "user.registered", user_id=user["id"])

    return {"id": user["id"]}           # ← returns NOW; email sends after

# Timeline:
#   0ms    request in → user created
#   15ms   201 response SENT — user's spinner is gone
#   15ms+  send_welcome_email runs (2s), then audit — user never waited
#
# Works inside dependencies too (e.g., a usage-metering dependency that
# add_tasks a counter bump for every call). Same BackgroundTasks object
# is shared between dependencies and the endpoint.

WebSockets — Real-Time, Two-Way Connections

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.rooms: dict[str, set[WebSocket]] = {}      # room → live sockets

    async def connect(self, room: str, ws: WebSocket):
        await ws.accept()                               # completes the handshake
        self.rooms.setdefault(room, set()).add(ws)

    def disconnect(self, room: str, ws: WebSocket):
        self.rooms.get(room, set()).discard(ws)         # ALWAYS clean up

    async def broadcast(self, room: str, message: dict):
        dead = []
        for ws in self.rooms.get(room, set()):
            try:
                await ws.send_json(message)
            except Exception:
                dead.append(ws)                         # died mid-broadcast
        for ws in dead:
            self.disconnect(room, ws)

manager = ConnectionManager()

@app.websocket("/ws/drives/{drive_id}/chat")
async def drive_chat(ws: WebSocket, drive_id: str):
    await manager.connect(drive_id, ws)
    await manager.broadcast(drive_id, {"sys": "someone joined"})
    try:
        while True:                                     # the connection's life
            data = await ws.receive_json()              # blocks until a message
            await manager.broadcast(drive_id, {
                "from": data.get("name", "anon"),
                "text": data["text"],
            })
    except WebSocketDisconnect:                         # tab closed, network died
        manager.disconnect(drive_id, ws)
        await manager.broadcast(drive_id, {"sys": "someone left"})

# Browser side:
#   const ws = new WebSocket("wss://api.example.com/ws/drives/7/chat")
#   ws.onmessage = (e) => render(JSON.parse(e.data))
#   ws.send(JSON.stringify({name: "Asha", text: "When is the TCS drive?"}))

Streaming & SSE — Pushing Data as It Happens

import csv, io
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

# ✗ The naive export: builds ALL rows in RAM, ships once
# @app.get("/export")  → 4 lakh students × 500B = 200MB string + timeout risk

# ✓ Streaming: constant memory, first bytes arrive immediately
@app.get("/students/export.csv")
async def export_students():
    async def rows():
        buf = io.StringIO()
        writer = csv.writer(buf)
        writer.writerow(["id", "name", "branch", "cgpa"])       # header
        yield buf.getvalue(); buf.seek(0); buf.truncate(0)

        async for student in iter_students_in_batches(1000):    # keyset pages!
            writer.writerow([student.id, student.name,
                             student.branch, student.cgpa])
            yield buf.getvalue(); buf.seek(0); buf.truncate(0)

    return StreamingResponse(
        rows(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=students.csv"},
    )
# Memory: one row's worth, forever. The DB pagination inside is the
# keyset pattern from the Databases chapters — OFFSET would re-scan.

# Same tool proxies big files without buffering:
#   return StreamingResponse(s3_object.iter_chunks(),
#                            media_type="application/pdf")