Connection Manager Pattern

~13 min · fastapi, manager, broadcast

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

One handler is not enough

An echo server only ever talks to the client that connected. Anything more interesting — chat, broadcast, presence — needs the server to know about multiple connections at once. The pattern is to introduce a connection manager: a small singleton that tracks open sockets, indexed by whatever your application cares about (room, user, channel).

Minimal viable manager

A dict mapping room name to a set of WebSocket objects covers most cases. Add user-id tracking when you need direct messaging. Add metadata (connected_at, ip) when you need observability. Each addition is incremental.

Always handle dead connections during broadcast

By the time you iterate over the room's set to broadcast, some sockets may already be half-dead (network drop, not yet detected). Every send call must be wrapped — collect dead refs, drop them after the loop, do not mutate the set during iteration.

Code

Minimal connection manager·python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, Set

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.rooms: Dict[str, Set[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        self.rooms.setdefault(room, set()).add(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        members = self.rooms.get(room)
        if not members:
            return
        members.discard(websocket)
        if not members:
            self.rooms.pop(room, None)  # garbage collect empty rooms

    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)
        for ws in dead:
            self.disconnect(ws, room)

manager = ConnectionManager()

@app.websocket('/ws/{room}')
async def chat(websocket: WebSocket, room: str):
    await manager.connect(websocket, room)
    try:
        async for msg in websocket.iter_json():
            await manager.broadcast(room, {
                'type': 'chat.message',
                'data': msg,
            })
    except WebSocketDisconnect:
        manager.disconnect(websocket, room)
        await manager.broadcast(room, {
            'type': 'user.left',
            'data': {'room': room},
        })

External links

FastAPI advanced: WebSockets

Exercise

Implement the manager above. Open three browser tabs to /ws/general. Type in one and verify the message appears in the other two. Crash one tab and verify the remaining two get a user.left broadcast, and that the room dict is cleaned up when the last tab closes.