Guía para principiantes de FastAPI × WebSocket: patrones de implementación para comunicación en tiempo real, chat y paneles (dashboards)
Resumen (primero, la visión general)
- WebSocket es un protocolo que permite comunicación bidireccional y en tiempo real entre servidor y cliente, lo que lo hace ideal para chat, notificaciones y actualizaciones en vivo de dashboards.
- FastAPI hereda capacidades de Starlette. Usando el decorador
@app.websocket()y la claseWebSocket, puedes implementar un servidor WebSocket sencillo. - Este artículo presenta patrones paso a paso: un WebSocket mínimo para un solo cliente, difusión (broadcast) a múltiples clientes y un “connection manager” basado en clases.
- También busca ser práctico para trabajo real incluyendo temas de seguridad (verificación de Origin, autenticación por token), pruebas de WebSockets (cómo testear con
TestClient) y una pequeña muestra de front-end (JavaScript). - Por último, resume puntos de diseño por caso de uso (chat, dashboards, notificaciones) y ofrece una hoja de ruta de “por dónde empezar”.
A quién le conviene leer esto (perfiles concretos)
Desarrolladores indie / estudiantes
- Quieres añadir algo como una “UI tipo chat” o un “contador en tiempo real” a tu servicio personal.
- Puedes construir APIs REST, pero todavía te resulta confuso “para qué sirven los WebSockets”.
- Quieres sentir la comunicación bidireccional tocando tanto JavaScript como FastAPI.
Para estas personas, construiremos hasta llegar a “algo que se mueve” combinando un endpoint WebSocket mínimo con un ejemplo simple tipo chat en HTML + JavaScript.
Ingenieros backend en equipos pequeños
- Tus pantallas de administración o dashboards necesitan “actualizaciones instantáneas” y el polling empieza a doler.
- Quieres incorporar elementos en tiempo real—notificaciones, chat, indicadores de progreso—en una app FastAPI existente.
- Quieres organizar la gestión de conexiones (detección de desconexión, broadcast) y una estrategia de testing.
Para estas personas, serán especialmente útiles patrones como un manager de conexiones, separación por salas (rooms) y cómo testear.
Equipos SaaS / startups
- Trabajas en áreas donde la inmediatez se convierte en valor de negocio: monitoreo, inventario en tiempo real, hubs de notificación, etc.
- Quieres un diseño que no te encierre, manteniendo en mente el escalado futuro (múltiples instancias, integración con broker externo).
- Quieres decidir autenticación, formatos de mensajes y diseño de salas de una manera extensible.
Para estas personas, partiremos de una base sólida de “instancia única” y compartiremos pistas de diseño que anticipen emparejarlo con Redis Pub/Sub o un broker de mensajes más adelante.
Evaluación de accesibilidad (legibilidad y comprensión)
- Arquitectura de la información
- Concepto de WebSocket → bases en FastAPI → ejemplo simple → connection manager → seguridad → testing → casos de uso, en pasos escalonados.
- Terminología
- Usa vocabulario cotidiano cuando es posible (conexión, mensaje, broadcast), minimizando términos en inglés y explicándolos en la primera aparición.
- Código
- Dividido en bloques pequeños para que ningún snippet se vuelva demasiado largo. Comentarios ligeros para evitar ruido visual.
- Suposiciones sobre el lector
- Asume que ya terminaste un tutorial básico de Python + FastAPI, intentando que cada capítulo se entienda también por sí solo.
En conjunto, la estructura está escrita para un rango amplio de lectores y apunta a una claridad tipo WCAG AA en la organización del texto.
1. ¿Qué es WebSocket? Contraste rápido con HTTP
Aclaremos brevemente “qué es realmente WebSocket”.
1.1 Características de WebSocket
WebSocket es un protocolo para comunicación duradera (long-lived), de ida y vuelta, entre el navegador y el servidor.
- Una vez establecida la conexión, tanto cliente → servidor como servidor → cliente pueden enviar mensajes cuando quieran.
- A diferencia de HTTP (“solicitud → respuesta y termina”), WebSocket es como reutilizar una sola conexión muchas veces.
- Se usa comúnmente para chat, juegos, notificaciones, dashboards—cualquier cosa que requiera actualizaciones en tiempo real.
Si intentas hacer esto con HTTP “plano”, por lo general necesitas “polling” (llamar a un API cada pocos segundos), lo cual aumenta tráfico innecesario y tiene límites reales en la respuesta en tiempo real.
1.2 FastAPI y WebSockets
FastAPI está construido sobre el framework ASGI Starlette, y los WebSockets vienen de la funcionalidad de Starlette.
- Endpoints HTTP:
@app.get()/@app.post()etc. - Endpoints WebSocket:
@app.websocket()
No es tan especial como suena—básicamente defines una función que recibe un objeto WebSocket “en lugar de HTTP”.
2. Endpoints WebSocket en FastAPI: lo básico
Veamos código real.
2.1 El servidor WebSocket más pequeño
Primero, el ejemplo mínimo: hablar con un solo cliente.
# app/main.py
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
# Aceptar la conexión del cliente
await websocket.accept()
while True:
# Recibir un mensaje (texto)
data = await websocket.receive_text()
# Devolverlo tal cual
await websocket.send_text(f"Echo: {data}")
Tres puntos clave:
- Define una ruta WebSocket con
@app.websocket("/ws/echo"). - Recibe un objeto
WebSocketcomo parámetro de la función. - Acepta la conexión con
await websocket.accept(), y luego usareceive_*/send_*.
Además de receive_text() / send_text(), también puedes usar métodos binarios como receive_bytes() / send_bytes(), o helpers JSON como receive_json() / send_json().
2.2 Un cliente simple (lado del navegador)
Después de arrancar el servidor con uvicorn app.main:app --reload, puedes probar esto en la consola del navegador (DevTools):
// En la consola del navegador
const ws = new WebSocket("ws://localhost:8000/ws/echo");
ws.onopen = () => {
console.log("connected");
ws.send("hello");
};
ws.onmessage = (event) => {
console.log("received:", event.data);
};
Si ves logs como:
connected
received: Echo: hello
…entonces diste tu primer paso en el mundo WebSocket.
3. Soportar múltiples clientes: fundamentos de un connection manager
En UIs en tiempo real, casi siempre tienes “múltiples clientes” compartiendo la misma información.
Aquí introducimos la idea de un “connection manager” que mantiene el registro de clientes activos.
3.1 Clase Connection Manager
# app/websocket_manager.py
from typing import List
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
active_connectionsalmacena objetosWebSocketconectados.broadcast()envía un mensaje a todos los clientes activos.
3.2 Un WebSocket tipo chat usando el connection manager
# app/main.py (extracto)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.websocket_manager import ConnectionManager
app = FastAPI()
manager = ConnectionManager()
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Message: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("a client disconnected")
Cuando ocurre WebSocketDisconnect, llamamos a disconnect() para quitarlo de la lista y notificar al resto que alguien se fue.
3.3 Un cliente HTML súper simple
Construyamos la UI de chat más pequeña usando solo HTML.
<!-- Ejemplo: static/chat.html -->
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8" />
<title>FastAPI WebSocket Chat</title>
<style>
body { font-family: sans-serif; }
#messages { border: 1px solid #ccc; height: 200px; overflow-y: scroll; padding: 4px; }
#input-area { margin-top: 8px; }
</style>
</head>
<body>
<h1>Chat</h1>
<div id="messages"></div>
<div id="input-area">
<input id="msg" type="text" placeholder="Message" />
<button id="send">Send</button>
</div>
<script>
const messages = document.getElementById("messages");
const input = document.getElementById("msg");
const sendBtn = document.getElementById("send");
const ws = new WebSocket("ws://localhost:8000/ws/chat");
ws.onmessage = (event) => {
const div = document.createElement("div");
div.textContent = event.data;
messages.appendChild(div);
messages.scrollTop = messages.scrollHeight;
};
sendBtn.onclick = () => {
if (input.value) {
ws.send(input.value);
input.value = "";
}
};
</script>
</body>
</html>
Incluso con esta estructura simple, abrir dos ventanas del navegador te permite experimentar “un chat en tiempo real donde los mensajes aparecen al instante en la otra ventana”.
4. Patrones prácticos comunes: salas, usuarios y formatos de mensaje
A continuación, “ordenamos” brevemente patrones que aparecen con frecuencia en chats/dashboards reales.
4.1 Separación por sala (room)
A menudo quieres salas, no un chat global abierto.
En ese caso, puedes gestionar conexiones con una estructura como dict[str, list[WebSocket]].
# app/websocket_manager_rooms.py
from collections import defaultdict
from typing import Dict, List
from fastapi import WebSocket, WebSocketDisconnect
class RoomManager:
def __init__(self):
self.rooms: Dict[str, List[WebSocket]] = defaultdict(list)
async def connect(self, room: str, websocket: WebSocket):
await websocket.accept()
self.rooms[room].append(websocket)
def disconnect(self, room: str, websocket: WebSocket):
self.rooms[room].remove(websocket)
if not self.rooms[room]:
del self.rooms[room]
async def broadcast(self, room: str, message: str):
for ws in self.rooms.get(room, []):
await ws.send_text(message)
Puedes recibir el nombre de la sala mediante la ruta del WebSocket o parámetros de query, y luego hacer broadcast solo a los clientes de esa sala.
4.2 Usar mensajes JSON
Si usas strings planos, se vuelve difícil crecer cuando aumentan los tipos de mensajes.
En la práctica, es común usar JSON con campos como type y payload.
import json
async def handle_message(data: str, websocket: WebSocket):
msg = json.loads(data)
msg_type = msg.get("type")
payload = msg.get("payload", {})
if msg_type == "chat":
text = payload.get("text", "")
# Enviar un mensaje de chat a todos
elif msg_type == "join":
# Manejar entrada a una sala
elif msg_type == "leave":
# Manejar salida de una sala
else:
await websocket.send_text("unknown message type")
Si el front-end también envía/recibe JSON, puedes añadir funciones más adelante sin cambiar demasiado el formato.
5. Combinar con seguridad y autenticación
Los WebSockets también tienen preocupaciones de seguridad. Aquí lo esencial.
5.1 Origin y una mentalidad tipo CORS
Los WebSockets en navegador, como HTTP, requieren pensar “qué orígenes están permitidos”.
- Restringir Origin vía Nginx o un reverse proxy
- O rechazar comprobando
websocket.headers(p. ej.,Origin,Sec-WebSocket-Protocol) del lado del servidor
Un enfoque práctico es pasar un token como parámetro de query al conectarse y autenticar en el servidor.
5.2 Combinar con JWT
Por ejemplo, pasa un JWT emitido por login HTTP en la query string del WebSocket.
// Idea de front-end
const token = "your-existing-JWT";
const ws = new WebSocket(`wss://example.com/ws/chat?token=${token}`);
En el servidor, puedes leer query params desde WebSocket en FastAPI.
# app/main.py (extracto)
from fastapi import WebSocket, WebSocketDisconnect
from app.core.jwt import decode_access_token # Supongamos una utilidad JWT de un artículo previo
@app.websocket("/ws/secure-chat")
async def websocket_secure_chat(websocket: WebSocket):
token = websocket.query_params.get("token")
if not token:
await websocket.close(code=1008) # Violación de política
return
try:
payload = decode_access_token(token)
except Exception:
await websocket.close(code=1008)
return
username = payload.get("sub", "anonymous")
await manager.connect(websocket)
await manager.broadcast(f"{username} joined")
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"{username}: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"{username} left")
Este ejemplo usa “token válido = OK” por simplicidad, pero en producción también considerarás scopes/roles, límites de conexión, etc.
6. Testing de WebSockets: cómo verificar con TestClient
La documentación oficial de FastAPI incluye una guía para testear WebSockets.
6.1 Sintaxis básica de test
Usa el mismo TestClient que para HTTP y llama client.websocket_connect().
# tests/test_websocket_echo.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_websocket_echo():
with client.websocket_connect("/ws/echo") as websocket:
websocket.send_text("hello")
data = websocket.receive_text()
assert data == "Echo: hello"
Dos puntos clave:
with client.websocket_connect("/ws/echo") as websocket:abre una conexión y la cierra automáticamente al salir del bloque.- La API del cliente ofrece
send_text()/receive_text()que reflejan el lado del servidor.
6.2 Probar broadcast
También puedes escribir tests con múltiples clientes.
def test_broadcast():
with client.websocket_connect("/ws/chat") as ws1:
with client.websocket_connect("/ws/chat") as ws2:
ws1.send_text("hello")
msg1 = ws1.receive_text()
msg2 = ws2.receive_text()
assert "hello" in msg1
assert "hello" in msg2
Tener tests ayuda a detectar regresiones rápidamente cuando:
- Cambias la lógica del connection manager
- Modificas el comportamiento de autenticación
7. Notas de diseño por caso de uso (tips rápidos)
Por último, algunos casos típicos y pistas de diseño.
7.1 Chat (1 a 1 / grupo)
- Gestiona conexiones por ID de sala (chat room).
- Usa mensajes JSON como
type=chat,payload={text, sender_id, room_id}. - Si solo usuarios autenticados pueden entrar, identifica al usuario vía JWT y verifica permiso para unirse a la sala.
7.2 Dashboards en tiempo real
- Agrega datos periódicamente en el backend y empuja actualizaciones solo cuando haya cambios.
- Si los datos son compartidos por todos, conecta a todos a una sola sala (p. ej.,
dashboard) y haz broadcast. - Si cada usuario tiene dashboard personalizado, crea “una sala por user ID” o filtra mensajes por user ID en el payload.
7.3 Notificaciones / streams de eventos
- Transmite eventos como nuevos comentarios, cambios de estado y alertas del sistema por WebSocket.
- Divide responsabilidades: WebSocket como “canal en tiempo real”, REST API como “recuperación de historial”.
- Si necesitas tracking de leído/no leído o replay tras reconexión, mantén WebSocket como “disparador” y vuelve a obtener los datos reales vía REST/DB para simplificar.
8. Hoja de ruta por tipo de lector: por dónde empezar
Desarrolladores indie / estudiantes
- Implementa el mínimo
@app.websocket("/ws/echo")y conéctate desde la consola del navegador. - Añade un connection manager y construye un chat simple basado en broadcast.
- Crea una pequeña UI de chat con HTML + JavaScript y prueba chatear entre dos ventanas.
- Cuando te sientas cómodo, cambia a mensajes JSON y transmite eventos más allá del chat.
Ingenieros backend en equipos pequeños
- Añade un endpoint WebSocket a una app FastAPI existente para “notificaciones” o un “dashboard”.
- Introduce un connection manager + gestión por salas para segmentar conexiones por usuario/proyecto.
- Integra con tu auth existente (JWT, etc.) para soportar “WebSockets solo autenticados”.
- Añade tests WebSocket con
TestClientpara proteger auth y broadcast.
Equipos SaaS / startups
- Construye un PoC: servidor WebSocket de una sola instancia (chat/dashboard) dentro de FastAPI.
- Diseña con abstracción para luego extraer el broadcast a un broker externo (Redis Pub/Sub, colas).
- Define una política de seguridad (auth, rate limiting, restricciones de Origin) y alinéala con la infraestructura (LB/reverse proxy).
- Añade monitoreo (conexiones, desconexiones, tasas de error) y logging (volumen de eventos) para convertirlo en un sistema operable.
9. Referencias (docs oficiales, artículos japoneses, etc.)
Nota: El contenido puede cambiar con el tiempo. Revisa cada sitio para la información más reciente.
-
Documentación oficial (FastAPI / Starlette)
-
Artículos / tutoriales en japonés
-
Artículos en inglés / ejemplos avanzados
Conclusión
- WebSocket es un protocolo para comunicación bidireccional en tiempo real—y FastAPI lo soporta de forma bastante simple.
- Con
@app.websocket()y la claseWebSocket, puedes empezar con un servidor echo mínimo, y luego añadir un connection manager y gestión de salas para construir chat multi-cliente o dashboards. - Pensando en seguridad (autenticación, restricciones de Origin) y agregando tests (WebSocket tests vía
TestClient), puedes seguir ampliando funciones con seguridad. - No necesitas apuntar a un sistema enorme desde el día uno. Empieza pequeño—chat o notificaciones—y ve familiarizándote gradualmente con el ecosistema WebSocket.
Si esto se convierte en el disparador para añadir un poco de “diversión” en tiempo real a tu app FastAPI, me alegrará mucho.
Prueba un paso a la vez, a un ritmo que te resulte cómodo.