Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis

💻 Código de Exemplo 01 — Endpoint WebSocket com Autenticação por Token na Query String

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query

app = FastAPI()

TOKENS_VALIDOS = {"abc123": "usuario_alice", "xyz789": "usuario_bob"}

@app.websocket("/ws/seguro")
async def websocket_autenticado(websocket: WebSocket, token: str = Query(...)) -> None:
    usuario = TOKENS_VALIDOS.get(token)
    if usuario is None:
        # Fecha antes de aceitar: nunca completa o handshake com credencial inválida.
        await websocket.close(code=1008)  # 1008 = Policy Violation
        return

    await websocket.accept()
    await websocket.send_text(f"Bem-vindo, {usuario}!")
    try:
        while True:
            mensagem = await websocket.receive_text()
            await websocket.send_text(f"[{usuario}] {mensagem}")
    except WebSocketDisconnect:
        print(f"{usuario} desconectou")

💻 Código de Exemplo 02 — ConnectionManager com Broadcast para Sala de Chat

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self) -> None:
        self.active_connections: dict[str, WebSocket] = {}

    async def connect(self, cliente_id: str, websocket: WebSocket) -> None:
        await websocket.accept()
        self.active_connections[cliente_id] = websocket

    def disconnect(self, cliente_id: str) -> None:
        self.active_connections.pop(cliente_id, None)

    async def broadcast(self, mensagem: str, remetente_id: str) -> None:
        for cliente_id, connection in self.active_connections.items():
            prefixo = "Você" if cliente_id == remetente_id else remetente_id
            await connection.send_text(f"{prefixo}: {mensagem}")

manager = ConnectionManager()

@app.websocket("/ws/sala/{cliente_id}")
async def sala_chat(websocket: WebSocket, cliente_id: str) -> None:
    await manager.connect(cliente_id, websocket)
    await manager.broadcast(f"{cliente_id} entrou na sala", remetente_id="sistema")
    try:
        while True:
            mensagem = await websocket.receive_text()
            await manager.broadcast(mensagem, remetente_id=cliente_id)
    except WebSocketDisconnect:
        manager.disconnect(cliente_id)
        await manager.broadcast(f"{cliente_id} saiu da sala", remetente_id="sistema")

💻 Código de Exemplo 03 — Broadcast Distribuído entre Réplicas via Redis Pub/Sub

import asyncio
import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from redis.asyncio import Redis

app = FastAPI()
redis = Redis.from_url("redis://localhost:6379/0", decode_responses=True)

CANAL_CHAT = "chat_geral"

class ConnectionManager:
    def __init__(self) -> None:
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket) -> None:
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket) -> None:
        self.active_connections.remove(websocket)

    async def enviar_local(self, mensagem: str) -> None:
        """Entrega apenas aos clientes conectados NESTA réplica do processo."""
        for connection in self.active_connections:
            await connection.send_text(mensagem)

manager = ConnectionManager()

async def escutar_redis_e_repassar() -> None:
    """Task de background: assina o canal Redis e repassa tudo que chega aos clientes locais."""
    pubsub = redis.pubsub()
    await pubsub.subscribe(CANAL_CHAT)
    async for mensagem in pubsub.listen():
        if mensagem["type"] == "message":
            await manager.enviar_local(mensagem["data"])

@app.on_event("startup")
async def iniciar_listener() -> None:
    asyncio.create_task(escutar_redis_e_repassar())

@app.websocket("/ws/chat")
async def chat_distribuido(websocket: WebSocket) -> None:
    await manager.connect(websocket)
    try:
        while True:
            texto = await websocket.receive_text()
            # Publica no Redis em vez de chamar broadcast local: todas as réplicas recebem,
            # inclusive esta, mantendo consistência entre múltiplos processos/servidores.
            await redis.publish(CANAL_CHAT, json.dumps({"mensagem": texto}))
    except WebSocketDisconnect:
        manager.disconnect(websocket)