Why WebSockets matter in modern backends
Building real-time features used to require complex polling loops or fragile external services. WebSockets in FastAPI: a practical Python Backend guide shows you how to handle bidirectional communication with just a few lines of code, letting your team ship live notifications, chat, collaborative editing, and live dashboards without reinventing the wheel. When you've shipped features that refresh every second via HTTP polling, you know the waste—WebSockets cut that friction dramatically.
Accept and loop — A minimal WebSocket endpoint
The simplest WebSocket server is almost embarrassingly straightforward. You define an endpoint, accept the connection, then loop to receive and send messages. Let's start there:
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def ws(socket: WebSocket):
await socket.accept()
while True:
msg = await socket.receive_text()
await socket.send_text(msg)This endpoint accepts a WebSocket connection and echoes every message back. On the client side (browser), you'd connect like this:
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => console.log(event.data);
ws.send("hello");That's it. No reconnection logic, no message queues, no infrastructure—just a handler that sits in a loop. In production, though, you'll hit two problems immediately: one client gets one message back, and when the client disconnects unexpectedly, your handler crashes. Let's handle disconnection gracefully:
@app.websocket("/ws")
async def ws(socket: WebSocket):
await socket.accept()
try:
while True:
msg = await socket.receive_text()
await socket.send_text(msg)
except Exception as e:
print(f"Client disconnected: {e}")
finally:
await socket.close()Now when the client hangs up—network glitch, browser tab closes, whatever—your handler exits cleanly instead of leaving a zombie coroutine behind.
Track connections — Keep a connection manager to broadcast to rooms
Real-time features rarely send a message back to one client. You want to broadcast: when Alice posts a comment, Bob and Carol see it instantly. You need to track active connections and route messages to them. Here's a connection manager pattern we've used across teams:
from typing import Set
class ConnectionManager:
def __init__(self):
self.active_connections: dict[str, Set[WebSocket]] = {}
async def connect(self, room_id: str, websocket: WebSocket):
await websocket.accept()
if room_id not in self.active_connections:
self.active_connections[room_id] = set()
self.active_connections[room_id].add(websocket)
def disconnect(self, room_id: str, websocket: WebSocket):
self.active_connections[room_id].discard(websocket)
async def broadcast(self, room_id: str, message: str):
if room_id in self.active_connections:
for connection in self.active_connections[room_id]:
try:
await connection.send_text(message)
except RuntimeError:
# Connection closed unexpectedly
pass
manager = ConnectionManager()
@app.websocket("/ws/{room_id}")
async def ws(websocket: WebSocket, room_id: str):
await manager.connect(room_id, websocket)
try:
while True:
msg = await websocket.receive_text()
await manager.broadcast(room_id, msg)
except Exception as e:
print(f"Error: {e}")
finally:
manager.disconnect(room_id, websocket)Now when client A sends a message to room chat-123, everyone in that room receives it. You've split connections by room, so you're not broadcasting to every client on the server—just the ones who care. This pattern scales your mental model from "one connection" to "many grouped connections." In a CRM system we worked on, teams used this to see live customer interactions; in a trading platform, it let multiple traders see order updates instantly.
Scale out — Use Redis pub/sub so every instance can reach every client
The connection manager works great on a single server. But the moment you deploy a second instance, clients connected to instance A never hear broadcasts from instance B. Redis pub/sub bridges that gap:
import redis.asyncio as redis
import json
class ConnectionManager:
def __init__(self, redis_client):
self.active_connections: dict[str, Set[WebSocket]] = {}
self.redis = redis_client
async def connect(self, room_id: str, websocket: WebSocket):
await websocket.accept()
if room_id not in self.active_connections:
self.active_connections[room_id] = set()
self.active_connections[room_id].add(websocket)
def disconnect(self, room_id: str, websocket: WebSocket):
self.active_connections[room_id].discard(websocket)
async def broadcast(self, room_id: str, message: str):
# Publish to Redis so other instances see it too
await self.redis.publish(f"room:{room_id}", message)
# Also send to local connections
if room_id in self.active_connections:
for connection in self.active_connections[room_id]:
try:
await connection.send_text(message)
except RuntimeError:
pass
async def listen_redis(self, room_id: str):
# Background task to listen for messages from other instances
pubsub = self.redis.pubsub()
await pubsub.subscribe(f"room:{room_id}")
async for message in pubsub.listen():
if message["type"] == "message":
if room_id in self.active_connections:
for connection in self.active_connections[room_id]:
try:
await connection.send_text(message["data"].decode())
except RuntimeError:
pass
@app.on_event("startup")
async def startup():
app.redis = await redis.from_url("redis://localhost")
app.manager = ConnectionManager(app.redis)
@app.websocket("/ws/{room_id}")
async def ws(websocket: WebSocket, room_id: str):
await app.manager.connect(room_id, websocket)
# Spin up a background listener for Redis messages
listen_task = asyncio.create_task(app.manager.listen_redis(room_id))
try:
while True:
msg = await websocket.receive_text()
await app.manager.broadcast(room_id, msg)
except Exception as e:
print(f"Error: {e}")
finally:
listen_task.cancel()
app.manager.disconnect(room_id, websocket)Now instance A publishes to Redis, instance B's background listener picks it up, and instance B delivers it to its clients. You've decoupled the publisher from the subscriber. In a ride-hailing system we worked on, drivers on instance A could broadcast their location; all passengers connected to instance B got updates in real time. Redis pub/sub is simple and battle-tested; it handles thousands of messages per second without breaking a sweat.
Authenticate — Validate a token during the handshake, not after
You don't want random users joining private rooms. Validate the client before you accept the connection:
from fastapi import Depends, Query
from jose import JWTError, jwt
def get_token(query_token: str = Query(...)):
return query_token
def verify_token(token: str = Depends(get_token)):
try:
payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
user_id = payload.get("sub")
if user_id is None:
raise JWTError("Invalid token")
return user_id
except JWTError:
raise
@app.websocket("/ws/{room_id}")
async def ws(
websocket: WebSocket,
room_id: str,
token: str = Query(...)
):
user_id = None
try:
payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
user_id = payload.get("sub")
if user_id is None:
await websocket.close(code=4001, reason="Invalid token")
return
except JWTError:
await websocket.close(code=4001, reason="Invalid token")
return
await app.manager.connect(room_id, websocket)
try:
while True:
msg = await websocket.receive_text()
# Optionally tag messages with user_id
await app.manager.broadcast(room_id, json.dumps({
"user": user_id,
"message": msg
}))
except Exception as e:
print(f"Error: {e}")
finally:
app.manager.disconnect(room_id, websocket)The token comes in via query string. You verify it before calling accept(). If it's invalid, you close the connection with a 4001 code (custom close code) and never open it. The client never gets access to the room. This is cleaner than accepting first and then finding out they're not allowed—there's no state to clean up, and the client gets immediate feedback.
Common mistakes
- Not handling disconnections. A client abruptly closes the tab. If you don't catch the exception, your handler crashes and the socket leaks. Always wrap your loop in try/finally.
- Blocking the event loop. WebSocket handlers run on the event loop. If you do
time.sleep(5)or a synchronous database query, you freeze all WebSocket connections on that instance. Useasyncio.sleep()and async drivers (asyncpg, motor, etc.). - Authenticating after accept. Once you accept a connection, the client is "in." Validating later is theater. Do it during the handshake or reject outright.
- Forgetting to unsubscribe from Redis. If you don't cancel the listen task or close the pubsub, you'll leak subscriptions. Use try/finally or context managers.
- Broadcasting to closed connections without catching exceptions. A connection closes between when you grab the set and when you try to send. You need a try/except around each send, or use
asyncio.gather(..., return_exceptions=True).
Key takeaways
- WebSocket endpoints in FastAPI are async functions that accept, loop, and handle messages—start with the simplest version and add complexity only when you need it.
- Use a connection manager to group clients by room and broadcast to them; this scales your mental model from one-to-one to one-to-many.
- Add Redis pub/sub when you run multiple instances; it lets any instance reach any client without direct knowledge of where they're connected.
- Authenticate during the WebSocket handshake (before accept), not after; you're validating access before granting it, not checking permissions on every message.
- Always handle disconnections gracefully with try/finally blocks; connection leaks will exhaust your file descriptors and crash the server.
- Never block the event loop in a WebSocket handler; use async libraries and await everything that touches I/O.
Comments
Loading comments…