import asyncio
from collections import defaultdict
from fastapi import WebSocket
class ConnectionManager:
def __init__(self):
self._rooms: dict[str, set[WebSocket]] = defaultdict(set)
self._lock = asyncio.Lock()
async def connect(self, room: str, websocket: WebSocket) -> None:
async with self._lock:
self._rooms[room].add(websocket)
async def disconnect(self, room: str, websocket: WebSocket) -> None:
async with self._lock:
members = self._rooms.get(room)
if members:
members.discard(websocket)
if not members:
self._rooms.pop(room, None)
async def broadcast(self, room: str, payload: dict) -> None:
async with self._lock:
targets = list(self._rooms.get(room, ()))
if not targets:
return
results = await asyncio.gather(
*(ws.send_json(payload) for ws in targets),
return_exceptions=True,
)
stale = [ws for ws, res in zip(targets, results) if isinstance(res, Exception)]
for ws in stale:
await self.disconnect(room, ws)
manager = ConnectionManager()
from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator
class ChatMessage(BaseModel):
author: str = Field(min_length=1, max_length=64)
body: str = Field(min_length=1, max_length=2000)
sent_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
@field_validator("body")
@classmethod
def strip_body(cls, value: str) -> str:
cleaned = value.strip()
if not cleaned:
raise ValueError("body cannot be blank")
return cleaned
def to_frame(self) -> dict:
return {
"type": "message",
"author": self.author,
"body": self.body,
"sent_at": self.sent_at.isoformat(),
}
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from pydantic import ValidationError
from .connection_manager import manager
from .schemas import ChatMessage
router = APIRouter()
@router.websocket("/ws/rooms/{room}")
async def chat_endpoint(websocket: WebSocket, room: str):
await websocket.accept()
await manager.connect(room, websocket)
await manager.broadcast(room, {"type": "system", "body": f"a user joined {room}"})
try:
while True:
raw = await websocket.receive_json()
try:
message = ChatMessage.model_validate(raw)
except ValidationError as exc:
await websocket.send_json({"type": "error", "detail": exc.errors()})
continue
await manager.broadcast(room, message.to_frame())
except WebSocketDisconnect:
pass
finally:
await manager.disconnect(room, websocket)
await manager.broadcast(room, {"type": "system", "body": "a user left"})
A WebSocket connection manager centralizes the bookkeeping that a real-time feature needs: tracking which sockets are open, grouping them by room, and fanning a single message out to every subscriber. This snippet shows the pattern with three collaborating pieces — the manager that owns the socket registry, a Pydantic message envelope, and the FastAPI endpoint that wires clients into a room.
In ConnectionManager, the core data structure is self._rooms, a mapping from a room name to a set of live WebSocket objects. A set is used rather than a list so that add and remove are O(1) and duplicate registration is harmless. Every mutation is guarded by an asyncio.Lock, because connect and disconnect can interleave with a broadcast running on another task; without the lock a socket could be closed and removed while the broadcast loop is still iterating over it, raising a RuntimeError: Set changed size during iteration.
The broadcast method snapshots the room's members under the lock, then sends outside the lock. This is deliberate: holding the lock across await ws.send_json(...) would serialize all sends and let one slow client block everyone else. Instead the sends are dispatched concurrently with asyncio.gather(..., return_exceptions=True), so a single dead connection surfaces as an exception in the results rather than aborting the whole fan-out. Dead sockets collected in stale are then reaped in a follow-up disconnect pass — a simple form of failure handling that keeps the registry from leaking closed connections.
ChatMessage in schemas defines the wire envelope with model_validate used to parse untrusted client input; malformed payloads raise ValidationError which the endpoint turns into a clean error frame instead of tearing down the socket.
chat routes shows the idiomatic FastAPI flow: await websocket.accept(), register with the manager, then loop on receive_json(). The WebSocketDisconnect exception is the normal termination path and simply triggers cleanup in the finally block. A pitfall worth noting is that this manager holds state in a single process, so it only broadcasts to clients on the same worker — scaling horizontally requires a shared backplane such as Redis pub/sub feeding each instance's broadcast.
Related snips
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
class Comment < ApplicationRecord
belongs_to :article
belongs_to :author, class_name: "User"
validates :body, presence: true, length: { maximum: 2_000 }Live comments with model broadcasts + turbo_stream_from
<%# private stream: turbo signs the serialized record name %>
<%= turbo_stream_from current_user %>
<section class="notifications">
<h1>Notifications</h1>Turbo Streams + authorization: signed per-user stream name
Share this code
Here's the card — post it anywhere.
