WebSockets con FastAPI y Redis

En este artículo implementaremos notificaciones en tiempo real...

from fastapi import FastAPI, WebSocket
import redis.asyncio as redis

app = FastAPI()
r = redis.Redis(host="localhost", port=6379)

@app.websocket("/ws/{channel}")
async def websocket_endpoint(ws: WebSocket, channel: str):
    await ws.accept()
    pubsub = r.pubsub()
    await pubsub.subscribe(channel)
    async for message in pubsub.listen():
        if message["type"] == "message":
            await ws.send_text(message["data"])

Este patrón permite escalar horizontalmente sin perder mensajes.