FastAPI WebSocket: endpoint, gestore di broadcast e test | We.Inc
Scrivi un endpoint WebSocket in FastAPI, invia a tutti i client con un connection manager, leggi i token dalla query, gestisci le disconnessioni e testa tutto con TestClient.
In FastAPI dichiari una route socket con @app.websocket("/ws") e una funzione async che riceve un WebSocket. Chiama await websocket.accept(), poi cicla su receive_text() e send_text(). Intercetta WebSocketDisconnect per fare pulizia quando un client se ne va. Installa prima il pacchetto websockets, e testa con TestClient.websocket_connect().
I metodi WebSocket che userai di più
Chiamata
Cosa fa
await websocket.accept()
Completa l'handshake; prima non si può inviare nulla
await websocket.receive_text()
Attende il prossimo messaggio di testo
await websocket.receive_json()
Attende un messaggio e lo interpreta come JSON
await websocket.send_text(s)
Invia un messaggio di testo
await websocket.send_json(obj)
Serializza un oggetto e lo invia
await websocket.close(code=1000)
Chiude la connessione con un codice di chiusura
Il generatore di endpoint qui sotto scrive un main.py funzionante a partire da poche scelte: il percorso della route, se i messaggi sono testo semplice o JSON, se ogni client riceve ogni messaggio, e se deve arrivare un token nella query string. Insieme arriva un file pytest corrispondente. Ogni opzione corrisponde a uno schema della documentazione FastAPI, riscritto qui come un unico file eseguibile.
Scegli le opzioni nel generatore e copia main.py in una cartella vuota.
Installa i pacchetti: pip install "fastapi[standard]" websockets, poi avvia l'app con fastapi dev main.py.
Connettiti dal WebSocket tester di We.Inc o dalla console del browser a ws://localhost:8000 più il tuo percorso.
Copia il file di test generato ed esegui pytest per verificare l'endpoint senza avviare un server.
Avvolgi il ciclo di ricezione in try ed except WebSocketDisconnect. Senza, ogni client che chiude la scheda lascia uno stack trace nei tuoi log.
Chiudere un socket prima di accept() fa rispondere il server all'handshake con HTTP 403, quindi il client non si apre mai. Accetta prima e poi chiudi con 1008 se vuoi che il client riceva un codice di violazione di policy leggibile.
Un elenco di connessioni in memoria conosce solo i client del processo corrente. Con più worker, ognuno invia a un sottoinsieme diverso.
Usa receive_json e send_json quando i messaggi portano più di un campo. Ti risparmiano una chiamata a json.loads e falliscono in modo evidente su input malformati.
I client browser non possono aggiungere un header Authorization a un socket, quindi leggi le credenziali da un parametro di query o da un cookie.
Domande frequenti
Perché il mio WebSocket FastAPI restituisce 404 o non fa l'upgrade?
Parti dall'installazione: la documentazione WebSocket di FastAPI chiede di aggiungere il pacchetto websockets insieme a FastAPI. Poi controlla il percorso, perché una route dichiarata con @app.websocket risponde solo agli upgrade socket, mai a semplici GET HTTP. Anche chiudere il socket prima di accept() si presenta come un handshake rifiutato (HTTP 403).
Come invio un messaggio a tutti i client?
Tieni i socket accettati in una lista dentro una piccola classe manager, aggiungili alla connessione, rimuovili alla disconnessione, e cicla sulla lista chiamando send_text. Il generatore qui sopra produce esattamente questo quando spunti l'opzione di broadcast.
Il broadcast funziona con più worker Uvicorn?
Non con una lista in memoria. Ogni worker è un processo separato con la propria lista. La documentazione FastAPI rimanda a encode/broadcaster, che inoltra i messaggi tramite Redis o PostgreSQL così ogni processo riceve ogni messaggio.
Posso usare Depends e Query su una route WebSocket?
Sì. Gli endpoint WebSocket accettano Depends, Query, Cookie, Header, Path e Security proprio come le route HTTP, quindi una dipendenza di autenticazione può essere condivisa tra entrambe.
Come testo un endpoint WebSocket?
Crea un TestClient per l'app e apri il socket in un blocco with usando client.websocket_connect("/ws"). Al suo interno, send_text e receive_text guidano la conversazione, e uscire dal blocco chiude la connessione.
Cosa fa il ciclo dell'endpoint
Un handler socket è una coroutine di lunga durata. Accetta una volta, poi resta su await receive_text() fino al messaggio successivo, quindi un singolo client inattivo non costa quasi nulla. Quando il client se ne va, la ricezione in attesa solleva WebSocketDisconnect, ed è il momento di fare pulizia.
Poiché l'handler è async, gli altri client continuano a essere serviti mentre uno è in attesa. Evita chiamate bloccanti come time.sleep o driver di database sincroni dentro il ciclo; bloccano ogni connessione del worker.
Autenticazione prima di accept
Leggi il token dalla query string come un normale parametro. Hai poi due modi per rifiutarlo. Chiudere prima di accept() si traduce in un HTTP 403 sull'handshake, che un browser riporta solo come connessione fallita. Accettare e poi chiudere con il codice 1008 permette al client di leggere il motivo e mostrare un invito ad accedere invece di riprovare all'infinito.
L'endpoint funzionante più piccolo
main.py: echo
Accetta, poi rimanda indietro ogni messaggio finché il client non si disconnette.
Broadcast con un connection manager
Una classe manager condivisa da ogni connessione
Funziona per un singolo processo. Sostituisci la lista con un relay basato su Redis prima di scalare.
Test con TestClient
Non serve un server: il client parla con l'app nello stesso processo. Eseguilo con pytest.
Generatore di endpoint WebSocket FastAPI
Scegli come deve comportarsi l'endpoint e copia un main.py completo più un file pytest che lo verifica.
Formato dei messaggi
Testo semplice
JSON
Invia ogni messaggio a tutti i client connessi
Richiedi un token nella query string (?token=...)
Esegui: fastapi dev main.py, poi connettiti a ws://localhost:8000{v}