python 41 lines · 9 steps

Building a WebSocket chat with FastAPI

A connection manager tracks live WebSockets and fans out every message to all connected clients.

Explained by highlit
FastAPI Intermediate websockets broadcast connection-management async real-time
1from fastapi import FastAPI, WebSocket, WebSocketDisconnect
2 
3app = FastAPI()
4 
5 
6class ConnectionManager:
7 def __init__(self) -> None:
8 self.active: dict[WebSocket, str] = {}
9 
10 async def connect(self, websocket: WebSocket, username: str) -> None:
11 await websocket.accept()
12 self.active[websocket] = username
13 await self.broadcast(f"* {username} joined the chat")
14 
15 def disconnect(self, websocket: WebSocket) -> str:
16 return self.active.pop(websocket, "someone")
17 
18 async def broadcast(self, message: str) -> None:
19 stale = []
20 for connection in self.active:
21 try:
22 await connection.send_text(message)
23 except RuntimeError:
24 stale.append(connection)
25 for connection in stale:
26 self.active.pop(connection, None)
27 
28 
29manager = ConnectionManager()
30 
31 
32@app.websocket("/ws/{room}")
33async def chat_room(websocket: WebSocket, room: str, username: str = "anon") -> None:
34 await manager.connect(websocket, username)
35 try:
36 while True:
37 text = await websocket.receive_text()
38 await manager.broadcast(f"[{room}] {username}: {text}")
39 except WebSocketDisconnect:
40 left = manager.disconnect(websocket)
41 await manager.broadcast(f"* {left} left the chat")
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A central registry of active connections is the backbone of any broadcast-style real-time app.
  2. 2Broadcasting must tolerate dead sockets — collect failures and prune them instead of crashing the loop.
  3. 3The endpoint's job is just to accept, loop, and clean up; the manager owns all shared state.

Related explainers

python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Building a WebSocket chat with FastAPI — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code