Especialização em Backend com Python & FastAPI • FastAPI & Python 3.12+ • Pydantic v2, SQLModel, Async/Await e Microsserviços

🗺️ Mapa Conceitual do Tópico

flowchart TD
    A["Cliente (Frontend / Mobile)"] -->|"1. GET /ws/live (Upgrade: websocket)"| B["FastAPI Endpoint (@app.websocket)"]
    B --> C{"Validação de Token / Origem"}
    
    C -->|"Token Inválido"| D["await websocket.close(code=1008)"]
    C -->|"Autenticado"| E["await websocket.accept()"]
    
    E --> F["ConnectionManager.connect(ws, canal_id)"]
    
    subgraph REALTIME["Loop de Comunicação Full-Duplex"]
        G["asyncio.gather: Enviar e Receber Concorrente"]
        H["websocket.receive_text() / json()"]
        I["websocket.send_text() / json()"]
        G --> H
        G --> I
    end
    
    F --> REALTIME

    subgraph ESCALA_REDIS["Escala Horizontal Multiprocesso"]
        J[("Redis Pub/Sub (Canal Global)")]
        K["Worker FastAPI Réplica A"] <--> J
        L["Worker FastAPI Réplica B"] <--> J
    end

    REALTIME <--> J

    H -->|"Queda de Conexão (WebSocketDisconnect)"| M["ConnectionManager.disconnect(ws)"]
    M --> N["Broadcast de Desconexão aos Demais Clientes"]

    style A fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style B fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    style C fill:#ede7f6,stroke:#7e57c2,stroke-width:2px
    style D fill:#ffebee,stroke:#e53935,stroke-width:2px
    style E fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style F fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style G fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style J fill:#fffde7,stroke:#fbc02d,stroke-width:2px
    style M fill:#ffebee,stroke:#e53935,stroke-width:2px
    style N fill:#e0f2f1,stroke:#009688,stroke-width:2px

🏛️ 1. O Protocolo WebSocket (RFC 6455) vs HTTP/REST

Endpoints REST tradicionais operam sob o modelo Half-Duplex Unidirecional: o cliente inicia a conexão com uma requisição HTTP, o servidor processa e devolve a resposta, e o canal TCP é encerrado ou mantido ocioso (keep-alive). Para cenários que exigem atualizações contínuas em tempo real (como chats corporativos, dashboards financeiros com cotações de alta frequência, monitoramento de telemetria IoT ou jogos multiplayer), o modelo de polling HTTP gera tráfego excessivo de cabeçalhos e latência intolerável.

O WebSocket estabelece uma conexão TCP persistente e bidirecional de baixíssimo overhead:

  1. Handshake Inicial: Inicia como uma requisição HTTP convencional contendo os cabeçalhos Upgrade: websocket e Connection: Upgrade.
  2. Upgrade de Protocolo: O servidor ASGI aceita a transição respondendo com status HTTP 101 Switching Protocols.
  3. Canal Full-Duplex: A partir deste instante, o socket TCP permanece aberto para envio e recebimento concorrente de frames leves de texto (JSON) ou binários sem o custo de novos cabeçalhos HTTP.

Códigos de Fechamento Padronizados (RFC 6455)

Ao encerrar uma conexão WebSocket com await websocket.close(code=...), deve-se utilizar códigos formais da especificação para informar o motivo ao cliente:

Código Nome na RFC Significado Arquitetural
1000 Normal Closure Fechamento intencional e regular da sessão (ex: logout do usuário).
1001 Going Away O servidor ou cliente está sendo desligado (ex: reload da página).
1008 Policy Violation Falha de autorização, violação de regras de segurança ou token expirado.
1011 Internal Error Exceção inesperada no servidor durante o processamento de uma mensagem.

🛡️ 2. Autenticação Segura em WebSockets

Uma das armadilhas mais comuns em WebSockets é que a API de WebSockets nativa dos navegadores (new WebSocket(url)) não permite adicionar cabeçalhos customizados (como Authorization: Bearer <token>).

As abordagens seguras recomendadas são:

  1. Token via Query Parameter: wss://api.empresa.com/ws?token=eyJhbG... (comum, mas pode vazar em logs de proxy se não for protegido por HTTPS/WSS estrito).
  2. Ticket Efêmero de Handshake: O cliente solicita um ticket único de vida curta (30 segundos) via REST autenticado e usa esse ticket para conectar no WebSocket.
  3. Autenticação na Primeira Mensagem: A conexão é aceita, mas fica em quarentena até que o cliente envie uma mensagem inicial { "tipo": "AUTH", "token": "..." }. Se o token não for recebido em 5 segundos, o socket é fechado com código 1008.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, status
from core.security import decodificar_token_jwt

app = FastAPI()

async def obter_usuario_websocket(websocket: WebSocket, token: str | None = Query(default=None)) -> str:
    if not token:
        # Rejeita antes do accept: fecha com Policy Violation
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        raise WebSocketDisconnect(code=status.WS_1008_POLICY_VIOLATION)
    
    usuario = decodificar_token_jwt(token)
    if not usuario:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        raise WebSocketDisconnect(code=status.WS_1008_POLICY_VIOLATION)
        
    return usuario

👥 3. O Padrão ConnectionManager Resiliente

Para gerenciar salas temáticas e múltiplos clientes conectados concorrentemente, encapsulamos as operações em um gerenciador de conexões:

from typing import Dict, List
from fastapi import WebSocket

class ConnectionManager:
    def __init__(self) -> None:
        # Mapeia sala_id para lista de conexões ativas naquela sala
        self.salas_ativas: Dict[str, List[WebSocket]] = {}

    async def conectar(self, sala_id: str, websocket: WebSocket) -> None:
        await websocket.accept()
        if sala_id not in self.salas_ativas:
            self.salas_ativas[sala_id] = []
        self.salas_ativas[sala_id].append(websocket)

    def desconectar(self, sala_id: str, websocket: WebSocket) -> None:
        if sala_id in self.salas_ativas:
            if websocket in self.salas_ativas[sala_id]:
                self.salas_ativas[sala_id].remove(websocket)
            if not self.salas_ativas[sala_id]:
                del self.salas_ativas[sala_id]

    async def broadcast_sala(self, sala_id: str, payload: dict) -> None:
        if sala_id in self.salas_ativas:
            # Envia para todas as conexões da sala; remove sockets mortos se falharem
            desconectados = []
            for conexao in self.salas_ativas[sala_id]:
                try:
                    await conexao.send_json(payload)
                except Exception:
                    desconectados.append(conexao)
            
            for morto in desconectados:
                self.desconectar(sala_id, morto)

manager = ConnectionManager()

🌐 4. Escala Horizontal com Redis Pub/Sub

Em ambientes de produção com contêineres e réplicas balanceadas por Nginx ou AWS ALB, dois usuários da mesma sala podem estar conectados em processos ou pods distintos. Um broadcast executado em memória afetará apenas os usuários conectados àquela réplica específica.

A solução canônica é o Redis Pub/Sub:

  • Cada instância da API assina os canais Redis das salas ativas.
  • Quando qualquer cliente envia uma mensagem, o FastAPI a publica no canal do Redis (redis.publish(canal, mensagem)).
  • Todas as réplicas recebem a mensagem do Redis e a retransmitem localmente para os clientes conectados em sua própria memória, garantindo sincronização global em tempo real.