WebSocket FastAPI : endpoint, gestionnaire de diffusion et tests | We.Inc
Écrivez un endpoint WebSocket FastAPI, diffusez à tous les clients avec un gestionnaire de connexions, lisez des tokens dans la query string, gérez les déconnexions et testez le tout avec TestClient.
Dans FastAPI, vous déclarez une route socket avec @app.websocket("/ws") et une fonction async qui reçoit un WebSocket. Appelez await websocket.accept(), puis bouclez sur receive_text() et send_text(). Interceptez WebSocketDisconnect pour faire le ménage quand un client part. Installez d'abord le paquet websockets, et testez avec TestClient.websocket_connect().
Les méthodes WebSocket que vous appellerez le plus
Appel
Rôle
await websocket.accept()
Termine le handshake ; rien ne peut être envoyé avant
await websocket.receive_text()
Attend le prochain message texte
await websocket.receive_json()
Attend un message et l'analyse en JSON
await websocket.send_text(s)
Envoie un message texte
await websocket.send_json(obj)
Sérialise un objet et l'envoie
await websocket.close(code=1000)
Termine la connexion avec un code de fermeture
Le générateur d'endpoint ci-dessous écrit un main.py fonctionnel à partir de quelques choix : le chemin de la route, des messages en texte brut ou en JSON, la diffusion de chaque message à tous les clients ou non, et l'obligation ou non de fournir un token dans la query string. Un fichier pytest correspondant l'accompagne. Chaque option correspond à un modèle de la documentation FastAPI, réécrit ici sous forme d'un seul fichier exécutable.
Choisissez les options dans le générateur et copiez main.py dans un dossier vide.
Installez les dépendances : pip install "fastapi[standard]" websockets, puis lancez l'app avec fastapi dev main.py.
Connectez-vous depuis le testeur WebSocket de We.Inc ou une console de navigateur sur ws://localhost:8000 suivi de votre chemin.
Copiez le fichier de test généré et lancez pytest pour vérifier l'endpoint sans démarrer de serveur.
Enveloppez la boucle de réception dans un try avec except WebSocketDisconnect. Sans cela, chaque client qui ferme son onglet laisse une stack trace dans vos logs.
Fermer un socket avant accept() fait répondre le serveur au handshake par un HTTP 403, et le client ne s'ouvre jamais. Acceptez d'abord, puis fermez avec 1008 si vous voulez que le client reçoive un code de violation de politique lisible.
Une liste de connexions en mémoire ne connaît que les clients du processus actuel. Lancez plusieurs workers et chacun diffusera à un sous-ensemble différent.
Utilisez receive_json et send_json dès que les messages contiennent plus d'un champ. Ils vous évitent un appel à json.loads et échouent clairement sur une entrée mal formée.
Les clients navigateur ne peuvent pas ajouter d'en-tête Authorization à un socket : lisez plutôt les identifiants dans un paramètre de requête ou un cookie.
Questions fréquentes
Pourquoi mon WebSocket FastAPI renvoie-t-il une 404 ou échoue-t-il à l'upgrade ?
Commencez par l'installation : la documentation WebSocket de FastAPI demande d'ajouter le paquet websockets en plus de FastAPI. Vérifiez ensuite le chemin, car une route déclarée avec @app.websocket ne répond qu'aux upgrades de socket, jamais aux GET HTTP classiques. Fermer le socket avant accept() se traduit aussi par un handshake refusé (HTTP 403).
Comment diffuser un message à tous les clients ?
Gardez les sockets acceptés dans une liste à l'intérieur d'une petite classe de gestion, ajoutez-les à la connexion, retirez-les à la déconnexion, et parcourez la liste en appelant send_text. Le générateur ci-dessus produit exactement cela quand vous cochez l'option de diffusion.
La diffusion fonctionne-t-elle avec plusieurs workers Uvicorn ?
Pas avec une liste en mémoire. Chaque worker est un processus distinct avec sa propre liste. La documentation FastAPI renvoie vers encode/broadcaster, qui relaie les messages via Redis ou PostgreSQL pour que chaque processus reçoive chaque message.
Puis-je utiliser Depends et Query sur une route WebSocket ?
Oui. Les endpoints WebSocket acceptent Depends, Query, Cookie, Header, Path et Security comme les routes HTTP, donc une dépendance d'authentification peut être partagée entre les deux.
Comment tester un endpoint WebSocket ?
Créez un TestClient pour l'app et ouvrez le socket dans un bloc with avec client.websocket_connect("/ws"). À l'intérieur, send_text et receive_text pilotent l'échange, et sortir du bloc ferme la connexion.
Ce que fait la boucle de l'endpoint
Un gestionnaire de socket est une coroutine de longue durée. Il accepte une fois, puis attend sur await receive_text() jusqu'au message suivant, si bien qu'un client inactif ne coûte presque rien. Quand le client s'en va, la réception en attente lève WebSocketDisconnect, et c'est le moment de faire le ménage.
Comme le gestionnaire est async, les autres clients continuent d'être servis pendant qu'un client attend. Évitez les appels bloquants comme time.sleep ou les drivers de base de données synchrones dans la boucle : ils figent toutes les connexions du worker.
Authentification avant accept
Lisez le token dans la query string comme un paramètre normal. Vous avez alors deux façons de le refuser. Fermer avant accept() se traduit par un HTTP 403 sur le handshake, qu'un navigateur signale seulement comme une connexion échouée. Accepter puis fermer avec le code 1008 permet au client de lire la raison et d'afficher une invitation à se connecter au lieu de réessayer indéfiniment.
Le plus petit endpoint fonctionnel
main.py : echo
Accepte, puis renvoie chaque message jusqu'à ce que le client se déconnecte.
Diffuser avec un gestionnaire de connexions
Une classe de gestion partagée par toutes les connexions
Fonctionne pour un seul processus. Remplacez la liste par un relais basé sur Redis avant de passer à l'échelle.
Tester avec TestClient
Aucun serveur nécessaire : le client communique avec l'app dans le même processus. Lancez-le avec pytest.
Générateur d'endpoint WebSocket FastAPI
Choisissez le comportement de l'endpoint et copiez un main.py complet ainsi qu'un fichier pytest qui le vérifie.
Format des messages
Texte brut
JSON
Envoyer chaque message à tous les clients connectés
Exiger un token dans la query string (?token=...)
Lancez : fastapi dev main.py, puis connectez-vous à ws://localhost:8000{v}