WebSockets and Streaming

7 pages · ~5 min✓ Reviewed

Builds on Testing and Deployment. Next up: Observability in Production.

FastAPIPage 1

WebSockets and Streaming: The Mental Model

1. Three Ways to Talk

	HTTP	WebSocket	SSE
Flow	One request, one response	Both ways, any time	Server to browser only
Connection	Per request	One upgraded TCP	Plain HTTP
Type	Any	Full-duplex	`text/event-stream`

WebSocket = one persistent connection, upgraded from HTTP
Browser EventSource reconnects automatically for SSE

2. Declaring Them

python

@app.websocket('/ws')
async def f(websocket: WebSocket):
    ...

StreamingResponse(gen(), media_type='text/plain')

gen() is a generator

Path, query, header, cookie params work like normal routes
No response_model, no status code on a WebSocket
StreamingResponse sends chunks as produced
Whole body is never held in memory

3. Uploads

UploadFile

A spooled temp file: small uploads stay in memory, large ones roll over to disk.

FastAPIPage 2

WebSocket Endpoint Workflow

1. Handshake first

Declare with @app.websocket('/ws')
await websocket.accept() before any send or receive
Skipping it breaks the handshake

Call order	Result
`accept()`, then send/receive	OK
send/receive first	RuntimeError, Uvicorn answers 500
`close()` before `accept()`	Handshake rejected, 403

2. Receive, send, close

Each receive_* awaits the next message
Echo loop: accept once, then receive and send forever

FastAPIPage 3

Disconnects and Broadcasting

1. Disconnects

Client leaves: the next receive raises WebSocketDisconnect
Wrap the loop in try/except WebSocketDisconnect
Clean up in except or finally, or dead sockets pile up and sends fail

2. Connection Manager

python

from fastapi import WebSocket
class ConnectionManager:
    active: list[WebSocket] = []; disconnect = lambda s, ws: ws in s.active and s.active.remove(ws)
    async def connect(self, ws): await ws.accept(); self.active.append(ws)
    async def broadcast(self, msg):
        for ws in list(self.active):
            try: await ws.send_text(msg)
            except Exception: self.disconnect(ws)

connect = accept + append; disconnect = remove

python

from fastapi import FastAPI, WebSocketDisconnect
app, manager = FastAPI(), ConnectionManager()
@app.websocket("/ws")
async def chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True: await manager.broadcast(await websocket.receive_text())
    except WebSocketDisconnect: manager.disconnect(websocket)

3. Safe Broadcast

Iterate a copy: list(self.active)
Removing while iterating skips entries

FastAPIPage 4

StreamingResponse and Server-Sent Events

1. Streaming a generator

StreamingResponse(gen(), media_type='text/plain') sends chunks as the generator yields them
Takes a sync or async generator; prefer async def + yield
A sync generator runs in a threadpool
time.sleep or blocking calls in an async generator freeze every request: use await asyncio.sleep

2. SSE wire format

Each event is data: <payload>\n\n
The blank line ends the event

Field	Use
`data:`	the payload
`event:`	event name
`id:`	last event id
`retry:`	reconnect delay (ms)

FastAPIPage 5

File Responses and UploadFile

1. FileResponse

FileResponse(path, filename='report.pdf') streams a file from disk
Sets Content-Length, Content-Type, Content-Disposition
Supports Range requests
Threat: ../../etc/passwd leaks files
Fix: resolve() the path, check it is inside base

python

@app.get("/files/{name}")
async def get_file(name: str):
    path = (BASE / name).resolve()
    if not path.is_relative_to(BASE):
        raise HTTPException(404)
    return FileResponse(path, filename="report.pdf")

BASE is your resolved base directory

2. UploadFile

file: UploadFile reads multipart form data; list[UploadFile] takes several
Both need python-multipart installed
Attributes: filename, content_type, file
Async methods: read(size), write(), seek(), close()
await file.read() on a huge file loads it all into RAM

python

async def up(file: UploadFile):
    with open(dest, "wb") as out:
        while chunk := await file.read(1024 * 1024):
            out.write(chunk)

read in 1 MB chunks

FastAPIPage 6

Cheat Sheet: Realtime and Files

1. WebSocket and broadcast

accept() first, then receive_text/bytes/json() and send_text/bytes/json()
Server close: close(code=1000); 1008 = failed auth

python

async def ws(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            await websocket.send_text(await websocket.receive_text())
    except WebSocketDisconnect:
        pass
    finally: manager.disconnect(websocket)

remove the socket in finally

Manager keeps an active list; loop over list(active)
Prune sockets whose send raises
Many workers: Redis pub/sub

2. Streams, SSE and files

	Call	Rule
Stream	`StreamingResponse(gen(), media_type=...)`	async generator

FastAPIPage 7

Check yourself

The endpoint's first line is data = await websocket.receive_text(), with no accept. What happens?

RuntimeError; Uvicorn answers the handshake with a 500.
Fix: await websocket.accept() first. close() before accept would give a 403.

Broadcast loops for ws in self.active: and calls self.active.remove(ws) when a send fails. What is the bug?

Removing while iterating skips the next socket.
Iterate list(self.active) and prune the failures.

An async generator behind an SSE endpoint calls time.sleep(1) between events. What breaks?

It blocks the event loop, so every request freezes.
Use await asyncio.sleep(1).

Text	`receive_text()`	`send_text(s)`
Bytes	`receive_bytes()`	`send_bytes(b)`
JSON	`receive_json()`	`send_json(data)`

	bytes = File()	UploadFile
Storage	all in memory	spooled to disk
Use for	tiny payloads	anything large

WebSockets and Streaming