FastAPI WebSocket: endpoint, gestor de difusión y tests | We.Inc
Escribe un endpoint WebSocket en FastAPI, difunde a todos los clientes con un gestor de conexiones, lee tokens de la query, gestiona las desconexiones y pruébalo todo con TestClient.
En FastAPI declaras una ruta de socket con @app.websocket("/ws") y una función async que recibe un WebSocket. Ejecuta await websocket.accept() y luego recorre en bucle receive_text() y send_text(). Captura WebSocketDisconnect para limpiar cuando un cliente se va. Instala antes el paquete websockets y haz las pruebas con TestClient.websocket_connect().
Los métodos de WebSocket que más usarás
Invocación
Qué hace
await websocket.accept()
Completa el handshake; no se puede enviar nada antes
await websocket.receive_text()
Espera el siguiente mensaje de texto
await websocket.receive_json()
Espera un mensaje y lo analiza como JSON
await websocket.send_text(s)
Envía un mensaje de texto
await websocket.send_json(obj)
Serializa un objeto y lo envía
await websocket.close(code=1000)
Termina la conexión con un código de cierre
El generador de endpoints de abajo escribe un main.py funcional a partir de unas pocas opciones: la ruta, si los mensajes son texto plano o JSON, si todos los clientes reciben todos los mensajes y si debe llegar un token en el query string. Incluye un archivo de pytest a juego. Cada opción corresponde a un patrón de la documentación de FastAPI, reescrito aquí como un único archivo ejecutable.
Elige las opciones en el generador y copia main.py en una carpeta vacía.
Instala las piezas: pip install "fastapi[standard]" websockets, y luego arranca la app con fastapi dev main.py.
Conéctate desde el probador de WebSocket de We.Inc o desde la consola del navegador en ws://localhost:8000 más tu ruta.
Copia el archivo de test generado y ejecuta pytest para comprobar el endpoint sin arrancar un servidor.
Envuelve el bucle de recepción en try y except WebSocketDisconnect. Sin eso, cada cliente que cierra su pestaña deja una traza de error en tus logs.
Cerrar un socket antes de accept() hace que el servidor responda al handshake con HTTP 403, así que el cliente nunca llega a abrirse. Acepta primero y luego cierra con 1008 si quieres que el cliente reciba un código de violación de política legible.
Una lista de conexiones en memoria solo conoce a los clientes del proceso actual. Si ejecutas varios workers, cada uno difunde a un subconjunto distinto.
Usa receive_json y send_json en cuanto los mensajes lleven más de un campo. Te ahorran un json.loads y fallan de forma clara con entradas mal formadas.
Los clientes de navegador no pueden añadir una cabecera Authorization a un socket, así que lee las credenciales de un parámetro de query o de una cookie.
Preguntas frecuentes
¿Por qué mi WebSocket de FastAPI devuelve 404 o no hace el upgrade?
Empieza por la instalación: la documentación de WebSocket de FastAPI te pide añadir el paquete websockets junto a FastAPI. Después revisa la ruta, porque una ruta declarada con @app.websocket solo responde a upgrades de socket, nunca a GET HTTP normales. Cerrar el socket antes de accept() también aparece como un handshake rechazado (HTTP 403).
¿Cómo difundo un mensaje a todos los clientes?
Guarda los sockets aceptados en una lista dentro de una pequeña clase gestora, añádelos al conectar, quítalos al desconectar y recorre la lista invocando send_text. El generador de arriba produce exactamente esto cuando marcas la opción de difusión.
¿La difusión funciona con varios workers de Uvicorn?
No con una lista en memoria. Cada worker es un proceso separado con su propia lista. La documentación de FastAPI remite a encode/broadcaster, que retransmite los mensajes a través de Redis o PostgreSQL para que todos los procesos reciban todos los mensajes.
¿Puedo usar Depends y Query en una ruta WebSocket?
Sí. Los endpoints WebSocket aceptan Depends, Query, Cookie, Header, Path y Security igual que las rutas HTTP, así que una dependencia de autenticación se puede compartir entre ambas.
¿Cómo pruebo un endpoint WebSocket?
Crea un TestClient para la app y abre el socket en un bloque with usando client.websocket_connect("/ws"). Dentro, send_text y receive_text dirigen la conversación, y al salir del bloque se cierra la conexión.
Qué hace el bucle del endpoint
Un manejador de socket es una corrutina de larga duración. Acepta una vez y luego se queda en await receive_text() hasta el siguiente mensaje, así que un cliente inactivo apenas cuesta nada. Cuando el cliente se va, la recepción pendiente lanza WebSocketDisconnect, y ese es tu momento para limpiar.
Como el manejador es async, los demás clientes siguen siendo atendidos mientras uno espera. Evita las operaciones bloqueantes como time.sleep o los drivers de base de datos síncronos dentro del bucle; congelan todas las conexiones del worker.
Autenticación antes de accept
Lee el token del query string como un parámetro normal. Después tienes dos formas de rechazarlo. Cerrar antes de accept() se convierte en un HTTP 403 en el handshake, que el navegador solo muestra como una conexión fallida. Aceptar y luego cerrar con el código 1008 permite al cliente leer el motivo y mostrar un aviso para iniciar sesión en lugar de reintentar sin fin.
El endpoint funcional más pequeño
main.py: eco
Acepta y luego devuelve cada mensaje como eco hasta que el cliente se desconecte.
Difusión con un gestor de conexiones
Una clase gestora compartida por todas las conexiones
Funciona en un solo proceso. Cambia la lista por un relé respaldado por Redis antes de escalar horizontalmente.
Pruebas con TestClient
No hace falta servidor: el cliente habla con la app dentro del mismo proceso. Ejecútalo con pytest.
Generador de endpoints WebSocket para FastAPI
Elige cómo debe comportarse el endpoint y copia un main.py completo más un archivo de pytest que lo comprueba.
Formato de los mensajes
Texto plano
JSON
Enviar cada mensaje a todos los clientes conectados
Exigir un token en el query string (?token=...)
Ejecuta: fastapi dev main.py, y luego conéctate a ws://localhost:8000{v}