FastAPI WebSocket Manager Fix
In this tutorial, you'll learn about FastAPI WebSocket Manager Fix. We cover key concepts, practical examples, and best practices.
The Problem
A single WebSocket connection is easy. Managing multiple concurrent connections — broadcasting, room membership, and disconnection cleanup — requires a connection manager.
Quick Fix
Wrong — no connection management
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
# Can't send to other connected clients
await websocket.send_text(f"Echo: {data}")
Output: Only echo functionality. No way to broadcast or manage rooms.
Correct — ConnectionManager
from fastapi import WebSocket, WebSocketDisconnect
from typing import List
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
async def send_personal(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"User says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("User disconnected")
Output: All connected clients receive messages. Cleanup happens on disconnect.
Room-based manager
class RoomManager:
def __init__(self):
self.rooms: dict[str, List[WebSocket]] = {}
async def connect(self, room: str, websocket: WebSocket):
await websocket.accept()
if room not in self.rooms:
self.rooms[room] = []
self.rooms[room].append(websocket)
def disconnect(self, room: str, websocket: WebSocket):
self.rooms[room].remove(websocket)
if not self.rooms[room]:
del self.rooms[room]
async def broadcast_to_room(self, room: str, message: str):
for ws in self.rooms.get(room, []):
await ws.send_text(message)
Output: Messages are scoped to rooms. Clients in different rooms don't see each other's messages.
Async generator for connection lifecycle
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_websocket(websocket: WebSocket):
await websocket.accept()
try:
yield websocket
finally:
print(f"Connection closed for {websocket.client}")
@app.websocket("/ws")
async def endpoint(websocket: WebSocket):
async with managed_websocket(websocket) as ws:
async for data in ws.iter_text():
await ws.send_text(f"Received: {data}")
Prevention
- Always use a ConnectionManager class for multi-client applications.
- Handle
WebSocketDisconnectto clean up stale connections. - Use room-based managers for chat, games, or any group-scoped communication.
Common Mistakes with websocket manager
- Mixing let bindings with <- bindings in do notation, producing type errors
- Overlapping type class instances that cause GHC to reject the program with ambiguous dispatch errors
- Non-exhaustive pattern matches that compile with warnings then crash at runtime
These mistakes appear frequently in real-world FASTAPI code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.
Practice Exercise
Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.
This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro