FastAPI WebSocket: endpoints, broadcasting en testen
FastAPI WebSocket
FastAPI WebSocket: endpoint, broadcastmanager en tests | We.Inc
Schrijf een FastAPI WebSocket-endpoint, broadcast naar elke client met een connection manager, lees tokens uit de query, vang disconnects af en test alles met TestClient.
In FastAPI declareert u een socketroute met @app.websocket("/ws") en een async functie die een WebSocket ontvangt. Roep await websocket.accept() aan en loop daarna over receive_text() en send_text(). Vang WebSocketDisconnect af om op te ruimen als een client vertrekt. Installeer eerst het websockets-pakket, en test met TestClient.websocket_connect().
De WebSocket-methoden die u het vaakst aanroept
Aanroep
Wat hij doet
await websocket.accept()
Rondt de handshake af; daarvoor kan niets worden verstuurd
await websocket.receive_text()
Wacht op het volgende tekstbericht
await websocket.receive_json()
Wacht op een bericht en parset het als JSON
await websocket.send_text(s)
Verstuurt een tekstbericht
await websocket.send_json(obj)
Serialiseert een object en verstuurt het
await websocket.close(code=1000)
Beëindigt de verbinding met een sluitcode
De endpointbouwer hieronder schrijft een werkende main.py op basis van een paar keuzes: het routepad, of berichten platte tekst of JSON zijn, of elke client elk bericht hoort, en of er een token in de query string moet zitten. Er hoort een bijpassend pytest-bestand bij. Elke optie komt overeen met een patroon uit de FastAPI-documentatie, hier herschreven als één bestand dat direct draait.
Kies de opties in de bouwer en kopieer main.py naar een lege map.
Installeer de onderdelen: pip install "fastapi[standard]" websockets, en start daarna de app met fastapi dev main.py.
Verbind vanuit de WebSocket-tester van We.Inc of een browserconsole met ws://localhost:8000 plus uw pad.
Kopieer het gegenereerde testbestand en draai pytest om het endpoint te controleren zonder een server te starten.
Zet de receive-lus in een try met except WebSocketDisconnect. Zonder dat laat elke client die zijn tabblad sluit een stack trace achter in uw logs.
Een socket sluiten vóór accept() laat de server de handshake beantwoorden met HTTP 403, waardoor de client nooit opent. Accepteer eerst en sluit dan met 1008 als u wilt dat de client een leesbare code voor een policy violation ontvangt.
Een lijst met verbindingen in het geheugen kent alleen de clients van het huidige proces. Draai meerdere workers en elk broadcast naar een andere deelverzameling.
Gebruik receive_json en send_json zodra berichten meer dan één veld bevatten. Ze besparen u een json.loads-aanroep en falen duidelijk bij misvormde invoer.
Browserclients kunnen geen Authorization-header aan een socket toevoegen, dus lees credentials in plaats daarvan uit een queryparameter of een cookie.
Veelgestelde vragen
Waarom geeft mijn FastAPI WebSocket een 404 of lukt de upgrade niet?
Begin bij de installatie: de FastAPI-documentatie over WebSockets vraagt u het websockets-pakket naast FastAPI toe te voegen. Controleer daarna het pad, want een route die met @app.websocket is gedeclareerd beantwoordt alleen socket-upgrades, nooit gewone HTTP-GET's. Ook de socket sluiten vóór accept() verschijnt als een geweigerde handshake (HTTP 403).
Hoe broadcast ik een bericht naar elke client?
Bewaar geaccepteerde sockets in een lijst binnen een kleine managerklasse, voeg ze toe bij verbinden, verwijder ze bij disconnect, en loop over de lijst met send_text. De bouwer hierboven genereert precies dit als u de broadcastoptie aanvinkt.
Werkt broadcasten met meerdere Uvicorn-workers?
Niet met een lijst in het geheugen. Elke worker is een apart proces met een eigen lijst. De FastAPI-documentatie verwijst naar encode/broadcaster, dat berichten via Redis of PostgreSQL doorgeeft zodat elk proces elk bericht hoort.
Kan ik Depends en Query gebruiken op een WebSocket-route?
Ja. WebSocket-endpoints accepteren Depends, Query, Cookie, Header, Path en Security net als HTTP-routes, dus een authenticatie-dependency kan tussen beide worden gedeeld.
Hoe test ik een WebSocket-endpoint?
Maak een TestClient voor de app en open de socket in een with-blok met client.websocket_connect("/ws"). Daarbinnen sturen send_text en receive_text het gesprek aan, en het blok verlaten sluit de verbinding.
Wat de endpointlus doet
Een sockethandler is een langlopende coroutine. Hij accepteert één keer en wacht daarna op await receive_text() tot het volgende bericht, dus een enkele client kost bijna niets als hij inactief is. Als de client vertrekt, gooit de lopende receive een WebSocketDisconnect, en dat is uw moment om op te ruimen.
Omdat de handler async is, worden andere clients gewoon bediend terwijl er één wacht. Vermijd blokkerende aanroepen zoals time.sleep of synchrone databasedrivers in de lus; die bevriezen elke verbinding in de worker.
Authenticatie vóór accept
Lees het token uit de query string als een gewone parameter. U hebt dan twee manieren om het te weigeren. Sluiten vóór accept() wordt een HTTP 403 op de handshake, die een browser alleen als mislukte verbinding meldt. Accepteren en daarna sluiten met code 1008 laat de client de reden lezen en een inlogscherm tonen, in plaats van eindeloos opnieuw te proberen.
Het kleinste werkende endpoint
main.py: echo
Accepteer en stuur daarna elk bericht terug tot de client de verbinding verbreekt.
Broadcasten met een connection manager
Een managerklasse die door elke verbinding wordt gedeeld
Werkt voor één proces. Vervang de lijst door een relay op basis van Redis voordat u opschaalt.
Testen met TestClient
Geen server nodig: de client praat binnen hetzelfde proces met de app. Draai het met pytest.
Bouwer voor FastAPI WebSocket-endpoints
Kies hoe het endpoint zich moet gedragen en kopieer een complete main.py plus een pytest-bestand dat het controleert.
Berichtformaat
Platte tekst
JSON
Elk bericht naar elke verbonden client sturen
Een token in de query string vereisen (?token=...)
Draai: fastapi dev main.py, en verbind dan met ws://localhost:8000{v}