FastAPI WebSocket: Endpoints, Broadcasting und Tests
FastAPI WebSocket
FastAPI WebSocket: Endpoint, Broadcast-Manager und Tests | We.Inc
Schreibe einen FastAPI-WebSocket-Endpoint, sende mit einem Connection Manager an alle Clients, lies Tokens aus der Query, behandle Verbindungsabbrüche und teste alles mit TestClient.
In FastAPI deklarierst du eine Socket-Route mit @app.websocket("/ws") und einer async-Funktion, die einen WebSocket entgegennimmt. Rufe await websocket.accept() auf und laufe dann in einer Schleife über receive_text() und send_text(). Fange WebSocketDisconnect ab, um aufzuräumen, wenn ein Client geht. Installiere vorher das Paket websockets und teste mit TestClient.websocket_connect().
Die WebSocket-Methoden, die du am häufigsten aufrufst
Aufruf
Was er tut
await websocket.accept()
Schließt den Handshake ab; vorher kann nichts gesendet werden
await websocket.receive_text()
Wartet auf die nächste Textnachricht
await websocket.receive_json()
Wartet auf eine Nachricht und parst sie als JSON
await websocket.send_text(s)
Sendet eine Textnachricht
await websocket.send_json(obj)
Serialisiert ein Objekt und sendet es
await websocket.close(code=1000)
Beendet die Verbindung mit einem Close-Code
Der Endpoint-Builder unten schreibt aus wenigen Entscheidungen eine funktionierende main.py: den Routenpfad, ob Nachrichten reiner Text oder JSON sind, ob jeder Client jede Nachricht hört und ob ein Token im Query-String ankommen muss. Eine passende pytest-Datei gibt es dazu. Jede Option entspricht einem Muster aus der FastAPI-Dokumentation, hier als eine einzige lauffähige Datei umgeschrieben.
Wähle die Optionen im Builder und kopiere main.py in einen leeren Ordner.
Installiere die Bausteine: pip install "fastapi[standard]" websockets, und starte die App dann mit fastapi dev main.py.
Verbinde dich über den WebSocket-Tester von We.Inc oder eine Browser-Konsole mit ws://localhost:8000 plus deinem Pfad.
Kopiere die erzeugte Testdatei und führe pytest aus, um den Endpoint zu prüfen, ohne einen Server zu starten.
Pack die Empfangsschleife in try und except WebSocketDisconnect. Ohne das hinterlässt jeder Client, der seinen Tab schließt, einen Stacktrace in deinen Logs.
Schließt du einen Socket vor accept(), beantwortet der Server den Handshake mit HTTP 403, und der Client öffnet nie. Akzeptiere zuerst und schließe dann mit 1008, wenn der Client einen lesbaren Code für eine Richtlinienverletzung erhalten soll.
Eine Liste von Verbindungen im Speicher kennt nur die Clients des aktuellen Prozesses. Laufen mehrere Worker, sendet jeder an eine andere Teilmenge.
Nutze receive_json und send_json, sobald Nachrichten mehr als ein Feld enthalten. Das spart dir einen json.loads-Aufruf und schlägt bei fehlerhaften Eingaben deutlich fehl.
Browser-Clients können einem Socket keinen Authorization-Header hinzufügen, lies Zugangsdaten also stattdessen aus einem Query-Parameter oder einem Cookie.
Häufige Fragen
Warum liefert mein FastAPI-WebSocket 404 oder schafft das Upgrade nicht?
Fang bei der Installation an: Die FastAPI-Doku zu WebSockets verlangt, dass du das Paket websockets zusätzlich zu FastAPI installierst. Prüfe dann den Pfad, denn eine mit @app.websocket deklarierte Route beantwortet nur Socket-Upgrades, nie normale HTTP-GETs. Auch ein Schließen des Sockets vor accept() zeigt sich als abgelehnter Handshake (HTTP 403).
Wie sende ich eine Nachricht an alle Clients?
Bewahre akzeptierte Sockets in einer Liste in einer kleinen Manager-Klasse auf, füge sie beim Verbinden hinzu, entferne sie beim Trennen und laufe über die Liste, um send_text aufzurufen. Der Builder oben erzeugt genau das, wenn du die Broadcast-Option anhakst.
Funktioniert Broadcasting mit mehreren Uvicorn-Workern?
Nicht mit einer Liste im Speicher. Jeder Worker ist ein eigener Prozess mit eigener Liste. Die FastAPI-Doku verweist auf encode/broadcaster, das Nachrichten über Redis oder PostgreSQL weiterleitet, sodass jeder Prozess jede Nachricht hört.
Kann ich Depends und Query in einer WebSocket-Route verwenden?
Ja. WebSocket-Endpoints akzeptieren Depends, Query, Cookie, Header, Path und Security genau wie HTTP-Routen, sodass eine Authentifizierungs-Dependency für beide genutzt werden kann.
Wie teste ich einen WebSocket-Endpoint?
Erstelle einen TestClient für die App und öffne den Socket in einem with-Block mit client.websocket_connect("/ws"). Darin steuern send_text und receive_text die Unterhaltung, und das Verlassen des Blocks schließt die Verbindung.
Was die Endpoint-Schleife tut
Ein Socket-Handler ist eine lang laufende Coroutine. Er akzeptiert einmal und wartet dann bei await receive_text() auf die nächste Nachricht, sodass ein einzelner Client im Leerlauf fast nichts kostet. Wenn der Client verschwindet, löst das ausstehende receive WebSocketDisconnect aus, und das ist dein Moment zum Aufräumen.
Weil der Handler async ist, werden andere Clients weiter bedient, während einer wartet. Vermeide blockierende Aufrufe wie time.sleep oder synchrone Datenbanktreiber in der Schleife; sie frieren jede Verbindung im Worker ein.
Authentifizierung vor accept
Lies das Token als normalen Parameter aus dem Query-String. Dann hast du zwei Wege, es abzulehnen. Ein Schließen vor accept() wird beim Handshake zu HTTP 403, was ein Browser nur als fehlgeschlagene Verbindung meldet. Akzeptieren und anschließend mit Code 1008 schließen lässt den Client den Grund lesen und eine Anmeldeaufforderung anzeigen, statt endlos neu zu versuchen.
Der kleinste funktionierende Endpoint
main.py: Echo
Akzeptieren und dann jede Nachricht zurückschicken, bis der Client die Verbindung trennt.
Broadcast mit einem Connection Manager
Eine Manager-Klasse für alle Verbindungen
Funktioniert für einen einzelnen Prozess. Tausche die Liste gegen eine Weiterleitung über Redis, bevor du horizontal skalierst.
Testen mit TestClient
Kein Server nötig: Der Client spricht im selben Prozess mit der App. Führe es mit pytest aus.
Builder für FastAPI-WebSocket-Endpoints
Lege fest, wie sich der Endpoint verhalten soll, und kopiere eine vollständige main.py plus eine pytest-Datei, die sie prüft.
Nachrichtenformat
Reiner Text
JSON
Jede Nachricht an alle verbundenen Clients senden
Ein Token im Query-String verlangen (?token=...)
Starten: fastapi dev main.py, dann mit ws://localhost:8000{v} verbinden