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 HTTP / Frontend"] --> B["API Gateway / Router"]
    B --> C["Controller / Handler"]
    C --> D["Service Layer (Regras de Negócio)"]
    D --> E["Repository / ORM (Persistência)"]
    E --> F["Banco de Dados / Cache"]

    subgraph ARQ["Arquitetura do Capítulo"]
        G["Conceito: Comunicação em Tempo Real Bidirecional com WebSockets"]
        H["Segurança, Validação e Resiliência"]
        I["Alta Performance e Escalabilidade"]
    end

    D --> ARQ

    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:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style E fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style F fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px

🏛️ 1. HTTP request-response vs. canal full-duplex

Todo endpoint HTTP tradicional segue o ciclo requisição→resposta: o cliente pergunta, o servidor responde, a conexão termina (ou fica ociosa em keep-alive). Para o servidor empurrar dados sem o cliente perguntar (cotações mudando, notificações, chat, dashboards ao vivo), a alternativa ingênua é polling repetido — desperdício de banda e latência perceptível. O WebSocket (ws:// ou wss:// sobre TLS) resolve isso fazendo um handshake HTTP inicial (Upgrade: websocket) e depois mantendo uma única conexão TCP aberta, bidirecional e full-duplex: qualquer lado pode enviar uma mensagem a qualquer momento, sem novo handshake.

No FastAPI, uma rota WebSocket usa o decorador @app.websocket(...) em vez de @app.get/@app.post, recebe um objeto WebSocket e segue um ciclo de vida explícito — aceitar, trocar mensagens em loop, tratar desconexão:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_echo(websocket: WebSocket) -> None:
    await websocket.accept()          # completa o handshake HTTP -> WS
    try:
        while True:
            texto = await websocket.receive_text()
            await websocket.send_text(f"Eco: {texto}")
    except WebSocketDisconnect:
        # o cliente fechou a aba ou caiu a conexão: sempre trate, ou o processo
        # tentará usar um socket morto na próxima iteração do loop.
        print("Cliente desconectado")

Um detalhe importante: Depends() funciona em rotas WebSocket para injetar dependências no momento da conexão (ex.: uma sessão de banco), mas o modelo de autorização é diferente do HTTP — não há “response” intermediária para devolver um 401 elegante; a prática correta é validar credenciais antes de accept() e, se inválidas, chamar await websocket.close(code=1008) (Policy Violation) sem nunca aceitar o handshake.

👥 2. ConnectionManager: broadcast para múltiplos clientes

Cada conexão WebSocket ativa é isolada por padrão. Para funcionalidades como chat ou notificações em grupo, o padrão é manter uma lista (ou dicionário) de conexões ativas em um objeto ConnectionManager, com métodos para registrar, remover e transmitir (broadcast):

from fastapi import WebSocket

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 broadcast(self, mensagem: str) -> None:
        for connection in self.active_connections:
            await connection.send_text(mensagem)

manager = ConnectionManager()

@app.websocket("/ws/chat/{cliente_id}")
async def chat_endpoint(websocket: WebSocket, cliente_id: str) -> None:
    await manager.connect(websocket)
    try:
        while True:
            mensagem = await websocket.receive_text()
            await manager.broadcast(f"{cliente_id}: {mensagem}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{cliente_id} saiu do chat")

🌐 3. Escalando horizontalmente: Pub/Sub com Redis

O ConnectionManager acima funciona apenas dentro de um único processo. Em produção, geralmente há múltiplas réplicas do servidor FastAPI atrás de um load balancer — se o cliente A está conectado à réplica 1 e o cliente B à réplica 2, um broadcast local nunca chegaria a B. A solução é desacoplar o broadcast da memória do processo usando Redis Pub/Sub (ou Redis Streams): cada réplica publica mensagens em um canal Redis e também assina esse canal; ao receber uma mensagem publicada por qualquer réplica (inclusive por ela mesma), repassa para seus próprios clientes WebSocket conectados localmente. Isso torna o sistema de tempo real horizontalmente escalável sem estado compartilhado direto entre processos.

🔗 Recursos Pedagógicos do Capítulo 17

Recurso Didático Finalidade Link de Acesso
📊 Slides de Aula Apresentação visual interativa com Dark Mode e suporte a teclado Ver Slides
🧠 Quiz Formativo Teste interativo de fixação com feedback imediato por alternativa Fazer Quiz
💻 Exemplos de Código Demonstrações funcionais com código executável Ver Exemplos
🧩 Exercícios em 4 Níveis Lista progressiva de fixação com gabarito em bloco colapsável Resolver Exercícios
⬅️ Capítulo Anterior 📚 Sumário de Tópicos Próximo Capítulo ➡️