backend / fastapi

WebSockets

Leçon 161 exercice

Explication

Ce que vous allez apprendre

Comprendre pourquoi une connexion WebSocket diffère fondamentalement du modèle requête/réponse HTTP
Ouvrir, utiliser et fermer proprement une connexion WebSocket avec FastAPI
Capturer WebSocketDisconnect pour nettoyer les ressources à la déconnexion d'un client
Construire un gestionnaire de connexions pour diffuser un message à plusieurs clients
Identifier la limite d'un gestionnaire de connexions en mémoire sur plusieurs instances de serveur

Dans quel contexte ?

Le produit doit ajouter un chat en temps réel pour la page de support client dans app/routers/chat.py : quand un agent répond, le message doit apparaître instantanément côté client, sans que ce dernier ait à rafraîchir la page ou interroger l'API en boucle (polling). Le modèle requête/réponse HTTP classique est mal adapté à ce besoin, puisque le serveur doit pouvoir pousser un message sans attendre une demande explicite du client — exactement ce que permet une connexion WebSocket persistante.

Pourquoi HTTP classique ne suffit pas toujours

Une requête HTTP classique suit un schéma "demande puis réponse". Le client demande, le serveur répond, la connexion se termine.

Ce modèle est mal adapté à des cas où le SERVEUR doit pouvoir envoyer des données sans être sollicité. Un chat en temps réel ou des notifications live, par exemple.

WebSocket répond exactement à ce besoin. Il établit une connexion PERSISTANTE et BIDIRECTIONNELLE entre le client et le serveur.

Une fois la connexion ouverte, après une "poignée de main" initiale basée sur HTTP, les deux côtés peuvent s'envoyer des messages à tout moment. Sans avoir à rouvrir une connexion à chaque échange — fondamentalement différent du modèle requête/réponse vu jusqu'ici dans ce cours.

Voyons maintenant le cycle de vie à respecter. await websocket.accept() ouvre officiellement la connexion.

Ensuite, une boucle while True attend et traite les messages entrants tant que la connexion reste ouverte. Un détail à ne jamais oublier : WebSocketDisconnect doit systématiquement être capturé.

C'est l'exception levée quand le client se déconnecte, en fermant l'onglet par exemple. C'est le bon endroit pour nettoyer les ressources associées, comme retirer le client d'une liste de connexions actives.

Une fois ce cycle maîtrisé, un vrai défi apparaît : gérer plusieurs connexions à la fois. Un seul WebSocket est simple, mais un chat implique de diffuser un message à TOUS les clients connectés à une même "room".

C'est le rôle d'un gestionnaire de connexions. Il garde la liste des WebSockets actifs et boucle dessus pour diffuser un message à tous.

Piège fréquent

Un gestionnaire de connexions basé sur une simple liste en mémoire fonctionne pour UNE seule instance de serveur, mais dès que l'application tourne sur plusieurs processus, un client connecté à une instance ne peut pas recevoir un broadcast déclenché sur une autre sans un mécanisme de synchronisation partagé comme Redis Pub/Sub.

Étape du cycle de vie	Méthode
Ouvrir la connexion	`await websocket.accept()`
Recevoir un message	`await websocket.receive_text()` / `receive_json()`
Envoyer un message	`await websocket.send_text()` / `send_json()`
Détecter une déconnexion	`except WebSocketDisconnect`

Commandes & code

WebSockets

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Écho : {data}")
    except WebSocketDisconnect:
        print("Client déconnecté")

# Gestionnaire de connexions multiples — broadcast à tous les clients connectés
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)


manager = ConnectionManager()


@app.websocket("/ws/chat/{room_id}")
async def chat_endpoint(websocket: WebSocket, room_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"[Room {room_id}] {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Un utilisateur a quitté la room {room_id}")

# Authentification sur une connexion WebSocket — via query param ou header au handshake
from fastapi import Query, status


@app.websocket("/ws/secure")
async def secure_websocket(websocket: WebSocket, token: Annotated[str, Query()]):
    try:
        payload = decode_access_token(token)
    except ValueError:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.accept()
    user_id = payload["sub"]

    try:
        while True:
            data = await websocket.receive_json()
            await websocket.send_json({"echo": data, "user_id": user_id})
    except WebSocketDisconnect:
        pass

# Échange JSON structuré avec Pydantic
from pydantic import BaseModel, ValidationError


class ChatMessage(BaseModel):
    content: str
    room_id: str


@app.websocket("/ws/typed-chat")
async def typed_chat(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            raw = await websocket.receive_json()
            try:
                message = ChatMessage.model_validate(raw)
            except ValidationError as e:
                await websocket.send_json({"error": e.errors()})
                continue

            await manager.broadcast(f"[{message.room_id}] {message.content}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

# Rooms multiples avec dictionnaire de connexions par room — scalabilité au sein d'un seul process
class RoomManager:
    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = {}

    async def join(self, room_id: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms.setdefault(room_id, []).append(websocket)

    def leave(self, room_id: str, websocket: WebSocket):
        self.rooms[room_id].remove(websocket)
        if not self.rooms[room_id]:
            del self.rooms[room_id]

    async def broadcast_to_room(self, room_id: str, message: str):
        for ws in self.rooms.get(room_id, []):
            await ws.send_text(message)


# NOTE production : pour scaler sur plusieurs instances/processes, un simple dict en
# mémoire ne suffit plus — il faut un pub/sub partagé (Redis Pub/Sub) entre les instances

Résumé

await websocket.accept() ouvre la connexion, receive_text/receive_json et send_text/send_json échangent des messages.
WebSocketDisconnect doit toujours être capturé pour nettoyer la connexion fermée côté serveur.
L'authentification WebSocket passe généralement par un token en query param (pas de header Authorization standard côté navigateur).
Au-delà d'une seule instance serveur, un pub/sub partagé (Redis) est nécessaire pour du broadcast cross-instances.

Exercices pratiques

1 disponible

Mission : le chat support qui plante à chaque déconnexion client

Objectif : Corriger un ConnectionManager de chat qui ne gère pas proprement la déconnexion des clients et sécuriser l'accès au WebSocket.

Contexte

Le chat support de app/routers/chat.py fonctionne bien tant que les clients restent connectés, mais dès qu'un client ferme son onglet, le serveur plante avec une exception non gérée et les autres clients de la room cessent de recevoir des messages. Par ailleurs, n'importe qui peut se connecter à /ws/chat/{room_id} sans authentification.

Résoudre l’exercice →