WebSockets e Comunicação em Tempo Real em Python (Ver se é duplicidade)

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>
            const ws  = new WebSocket("ws://localhost:8000/ws/echo");
            const log = document.getElementById("log");

            ws.onmessage = (e) => {
                log.innerHTML += `<p>← ${e.data}</p>`;
            };

            function enviar() {
                const msg = document.getElementById("msg").value;
                ws.send(msg);
                log.innerHTML += `<p>→ ${msg}</p>`;
            }
        </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

@app.get("/chat/{sala}", response_class=HTMLResponse)
async def pagina_chat(sala: str):
    return f"""
    <!DOCTYPE html>
    <html lang="pt-BR">
    <head>
        <meta charset="UTF-8">
        <title>Chat — {sala}</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}</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 ws       = new WebSocket(`ws://localhost:8000/ws/chat/{sala}/${{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";
                    div.innerHTML = `<span><b>${{propria ? "Você" : dados.usuario}}</b>:
                                     ${{dados.mensagem}}
                                     <small>${{dados.hora}}</small></span>`;
                }}

                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>
    """

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 fastapi import WebSocket, WebSocketDisconnect, Query
from jose import JWTError, jwt


async def autenticar_ws(
    websocket: WebSocket,
    token: str = Query(...)
) -> dict | None:
    """Autentica conexão WebSocket via token JWT na query string."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except JWTError:
        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.")

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


@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):
        self._conexoes:     Dict[str, WebSocket]       = {}
        self._historico:    Dict[str, List[Notificacao]] = {}

    async def registrar(self, usuario_id: str, websocket: WebSocket):
        await websocket.accept()
        self._conexoes[usuario_id]  = 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):
        self._conexoes.pop(usuario_id, None)
        print(f"Usuário {usuario_id} desconectado.")

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

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

    async def broadcast_todos(self, notificacao: Notificacao):
        for usuario_id in list(self._conexoes.keys()):
            await self.notificar(usuario_id, 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)


@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"}

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

Resumo

WebSockets estabelecem canal bidirecional persistente — ideal para comunicação em tempo real
FastAPI suporta WebSockets nativamente com @app.websocket() e WebSocketDisconnect
O GerenciadorConexoes centraliza broadcast e controle de salas de forma thread-safe
SSE é mais simples que WebSockets quando a comunicação é apenas servidor → cliente
Autenticação em WebSockets é feita via query string ou subprotocolo — não em headers
BackgroundTasks envia notificações WebSocket sem bloquear endpoints HTTP
Para produção com múltiplos workers, use Redis Pub/Sub para sincronizar mensagens entre processos

Referências e Leituras Complementares

FastAPI — WebSockets — https://fastapi.tiangolo.com/advanced/websockets/
MDN — WebSocket API — https://developer.mozilla.org/en-US/docs/Web/API/WebSocket
MDN — Server-Sent Events — https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
websockets — biblioteca Python standalone — https://websockets.readthedocs.io/en/stable/
RFC 6455 — protocolo WebSocket — https://datatracker.ietf.org/doc/html/rfc6455
BANKS, Alex; PORCELLO, Eve. Learning React. 2. ed. O'Reilly Media, 2020. — contexto de frontend para consumir WebSockets e SSE.
GRINBERG, Miguel. Flask Web Development. 2. ed. O'Reilly Media, 2018. — base para comparar abordagens de tempo real entre Flask e FastAPI.

WebSockets e Comunicação em Tempo Real em Python (Ver se é duplicidade) Já leu