Dashboards em tempo real são aplicações que mostram dados mudando continuamente, como métricas de CPU, pedidos chegando, sensores ou eventos de uso. A tecnologia central para esse tipo de experiência é o WebSocket, um canal de comunicação bidirecional que mantém uma conexão aberta entre navegador e servidor para troca rápida de mensagens.
Em FastAPI, WebSockets permitem construir painéis rápidos e reativos, mas a confiabilidade depende de decisões pequenas e consistentes. Boas práticas como autenticação antes de aceitar a conexão, isolamento por “salas”, controle de clientes lentos, bateladas de atualização e escalabilidade com Redis reduzem travamentos, vazamentos de memória e inconsistências entre instâncias.
Fundamentos: FastAPI, WebSocket e organização do projeto
FastAPI é um framework Python para APIs que também suporta WebSockets de forma nativa. Um endpoint WebSocket é uma rota especial que negocia o “upgrade” HTTP e passa a trocar frames continuamente. A conexão exige um accept do servidor, e geralmente fica em um loop de leitura e escrita. Uma boa base separa utilitários de envio, gerenciamento de conexões e regras de domínio.
Um projeto simples pode começar com um único arquivo, mas dashboards crescem rápido e se beneficiam de módulos. Um padrão funcional é ter um arquivo de aplicação, um módulo de utilidades de WebSocket e, se necessário, um módulo de barramento com Redis. As receitas seguintes assumem uma base comum, onde conexões são aceitas e tratadas com desligamento seguro. Esse cuidado inicial reduz duplicação e erros em produção.
pip install fastapi uvicorn[standard] pydantic redisfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()Receita 1: Eco mínimo e utilitário de envio seguro (contrato básico)
O primeiro passo é validar a fiação do WebSocket com um “eco”, em que o servidor devolve o que recebe. Isso cria um contrato simples e previsível: aceitar, ler mensagens, responder e lidar com desconexões. Em ambientes reais, falhas de rede, fechamento simultâneo e clientes interrompidos acontecem o tempo todo. Um utilitário de envio “seguro” evita que exceções derrubem tarefas e loops de broadcast.
O envio seguro costuma encapsular serialização e exceções em um único lugar. O uso de JSON como formato padrão facilita compatibilidade com navegadores e versionamento de payloads. Mesmo quando a exceção é ignorada, a conexão pode ser removida de conjuntos no gerenciamento de salas. Com isso, o código de negócio fica mais limpo e confiável.
# websocket_utils.py
import json
from fastapi import WebSocket
async def safe_send_json(ws: WebSocket, payload: dict) -> None:
try:
await ws.send_text(json.dumps(payload, ensure_ascii=False))
except Exception:
# Rede sumiu, corrida com close, ou cliente lento; ignora aqui
pass# app.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws/echo")
async def ws_echo(ws: WebSocket):
await ws.accept()
try:
while True:
msg = await ws.receive_text()
await ws.send_text(msg)
except WebSocketDisconnect:
passReceita 2: Autenticação na borda (token assinado ou cookie)
Autenticar “na borda” significa validar credenciais antes de aceitar a conexão. Em WebSocket, aceitar cedo demais permite conexões desnecessárias e aumenta consumo de recursos em cenários de abuso. Um padrão comum usa um JWT (token assinado) no parâmetro de query, validado antes de accept. Outra opção frequente é cookie de sessão, lido em ws.cookies e validado no servidor.
O JWT é um token que carrega declarações, chamadas claims, assinadas com um segredo do servidor. A validação verifica assinatura, expiração e campos esperados, retornando um dicionário com os dados. A partir disso, informações como usuário e permissões podem ser colocadas no contexto da conexão. Essa validação precoce reduz “ida e volta” e impede que conexões inválidas entrem em salas.
# app.py
from fastapi import FastAPI, WebSocket, Query, HTTPException
import jwt
app = FastAPI()
SEGREDO_JWT = "trocar-por-um-segredo-forte"
def verificar_jwt(token: str) -> dict:
try:
return jwt.decode(token, SEGREDO_JWT, algorithms=["HS256"])
except Exception:
raise HTTPException(status_code=401, detail="token inválido")
@app.websocket("/ws/metrics")
async def ws_metrics(ws: WebSocket, token: str = Query(...)):
claims = verificar_jwt(token)
await ws.accept(subprotocol="json")
# claims pode ser associado ao contexto da conexão conforme a necessidade
await ws.send_json({"type": "auth", "status": "ok", "sub": claims.get("sub")})# exemplo de leitura de cookie de sessão dentro do endpoint
@app.websocket("/ws/metrics-cookie")
async def ws_metrics_cookie(ws: WebSocket):
sessao = ws.cookies.get("session")
if not sessao:
# Sem cookie, não aceita a conexão
return
await ws.accept()Receita 3: Salas e tópicos (isolamento por dashboard e controle)
“Salas” ou “tópicos” são chaves que agrupam conexões interessadas nos mesmos dados. Em um dashboard, isso separa equipes, projetos, painéis ou filtros, evitando enviar atualizações para quem não precisa. O armazenamento costuma ser um dicionário de conjuntos: sala → conjunto de WebSockets. Essa estrutura é simples e muito eficiente para fan-out local.
Ao adicionar uma conexão na sala, o loop pode receber mensagens do cliente e, opcionalmente, repassar pequenos eventos de interface. Ao remover, é importante fazer o descarte no finally para garantir limpeza mesmo em exceções. Também é útil apagar a sala vazia para evitar crescimento de chaves inúteis. Esse isolamento facilita aplicar regras de permissão e limites por sala.
# app.py
from collections import defaultdict
from typing import Dict, Set
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from websocket_utils import safe_send_json
app = FastAPI()
salas: Dict[str, Set[WebSocket]] = defaultdict(set)
@app.websocket("/ws/room/{room_id}")
async def ws_room(ws: WebSocket, room_id: str):
await ws.accept()
salas[room_id].add(ws)
try:
while True:
data = await ws.receive_json()
for peer in list(salas[room_id]):
if peer is not ws:
await safe_send_json(peer, {"type": "event", "data": data})
except WebSocketDisconnect:
pass
finally:
salas[room_id].discard(ws)
if not salas[room_id]:
salas.pop(room_id, None)Receita 4: Fan-out no servidor (broadcast centralizado e seguro)
Fan-out é o ato de publicar uma atualização para várias conexões ao mesmo tempo. Em dashboards, a maior parte do tráfego é “servidor → cliente”, com o servidor empurrando deltas e medições. Centralizar o broadcast evita duplicação e padroniza o formato das mensagens. Também permite inserir métricas, logs e políticas em um único ponto.
Ao iterar um conjunto de WebSockets, desconexões podem ocorrer durante o envio e alterar o conjunto. Por isso, a iteração deve usar uma cópia, como list(...), evitando erro de mutação durante o loop. Em conjunto com envio seguro, isso impede que um cliente instável comprometa a publicação para os demais. O broadcast pode ser usado por tarefas periódicas, consumidores de fila e handlers de eventos.
# app.py
from websocket_utils import safe_send_json
async def broadcast(room_id: str, payload: dict) -> None:
for ws in list(salas.get(room_id, ())):
await safe_send_json(ws, payload)
# exemplo de uso:
# await broadcast("board:alpha", {"type": "delta", "rows": [...], "ts": 123.45})Receita 5: Heartbeats e timeouts de ociosidade (detectar conexões “fantasmas”)
Heartbeat é um sinal periódico para confirmar que a conexão continua viva. Navegadores suspendem abas, redes alternam entre Wi‑Fi e 4G e proxies encerram conexões ociosas. Sem heartbeat, o servidor pode manter conexões “fantasmas” em memória, acumulando salas e filas. Um padrão simples é o servidor enviar “ping” e esperar “pong” dentro de um prazo.
O timeout pode ser implementado com asyncio.wait_for, encerrando quando a resposta não chega. Em falha, a conexão é liberada e o cliente reconecta por conta própria, mantendo o dashboard consistente. É útil manter as mensagens de controle com um campo type, diferenciando pings de deltas. Esse mecanismo também ajuda a detectar clientes travados que não consomem mensagens.
# app.py
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws/ping")
async def ws_ping(ws: WebSocket):
await ws.accept()
try:
while True:
await ws.send_json({"type": "ping"})
_ = await asyncio.wait_for(ws.receive_text(), timeout=25)
# resposta esperada: "pong"
except (WebSocketDisconnect, asyncio.TimeoutError):
passReceita 6: Backpressure com filas (evitar que cliente lento derrube o fluxo)
Backpressure é o controle de pressão quando consumidores não conseguem acompanhar produtores. Em WebSocket, um cliente lento pode bloquear o loop de envio se o código tentar enviar diretamente para todos. Um padrão robusto é criar uma fila por cliente e um “sender task” que drena essa fila. Assim, o broadcast enfileira rapidamente e não fica preso em um único envio demorado.
A fila deve ter tamanho máximo, pois filas ilimitadas viram vazamento de memória em picos de eventos. Ao estourar, existem duas políticas comuns: descartar mensagens antigas ou encerrar a conexão lenta. Encerrar com código 1008 indica violação de política, útil para sinalizar “cliente lento”. Esse desenho torna o sistema previsível sob carga, mesmo com muitas abas abertas e redes instáveis.
# app.py
import asyncio
from asyncio import Queue, create_task
from typing import Dict
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from websocket_utils import safe_send_json
app = FastAPI()
MAX_FILA = 100
class Cliente:
def __init__(self, ws: WebSocket):
self.ws = ws
self.fila: Queue[dict] = Queue(MAX_FILA)
self.tarefa_envio = create_task(self._drenar())
async def _drenar(self) -> None:
while True:
payload = await self.fila.get()
await safe_send_json(self.ws, payload)
clientes: Dict[WebSocket, Cliente] = {}
async def tentar_enfileirar(ws: WebSocket, payload: dict) -> None:
c = clientes.get(ws)
if not c:
return
try:
c.fila.put_nowait(payload)
except asyncio.QueueFull:
await ws.close(code=1008)
@app.websocket("/ws/slow-safe")
async def ws_slow_safe(ws: WebSocket):
await ws.accept()
clientes[ws] = Cliente(ws)
try:
while True:
_ = await ws.receive_text()
except WebSocketDisconnect:
pass
finally:
c = clientes.pop(ws, None)
if c:
c.tarefa_envio.cancel()Receita 7: Bateladas e coalescência de deltas (menos mensagens, mais estabilidade)
Atualizações muito frequentes aumentam CPU, GC no navegador e overhead de JSON. Em dashboards, muitos eventos podem ser resumidos em uma atualização a cada 200–500 ms, mantendo sensação de “tempo real” sem saturar o sistema. Batching agrupa eventos e envia em lote, enquanto coalescência substitui várias mudanças por uma versão mais recente. O objetivo é enviar menos frames e reduzir contenção no loop assíncrono.
Uma implementação simples armazena eventos em um buffer e publica quando a janela de tempo expira. O tempo pode ser medido pelo relógio do event loop, e o buffer reinicia após o envio. Em gráficos, o payload pode conter pontos agregados ou somente os últimos valores por série. Esse padrão costuma reduzir picos e evitar travamentos quando a fonte gera “rajadas”.
# app.py
import asyncio
from typing import Any, AsyncIterator
async def publicador_coalescido(room_id: str, fonte: AsyncIterator[Any], janela_s: float = 0.25):
buffer_eventos = []
ultimo = asyncio.get_event_loop().time()
async for evt in fonte:
buffer_eventos.append(evt)
agora = asyncio.get_event_loop().time()
if agora - ultimo >= janela_s:
await broadcast(room_id, {"type": "delta", "events": buffer_eventos})
buffer_eventos = []
ultimo = agoraReceita 8: Contratos de schema com Pydantic e versionamento de payload
Um schema é uma estrutura definida para mensagens, com tipos e campos esperados. Em tempo real, schemas claros evitam erros silenciosos, pois clientes e servidor evoluem em ritmos diferentes. Pydantic valida e serializa dados, garantindo que números, strings e listas estejam no formato correto. O versionamento no payload permite manter compatibilidade quando o formato muda.
Um padrão seguro inclui campos como type e version, além do conteúdo do delta. Ao publicar, o modelo é transformado em dicionário com model_dump e então enviado. Com isso, o mesmo servidor pode suportar versões diferentes do dashboard por um período. A previsibilidade do contrato reduz exceções no front-end e facilita auditoria do tráfego.
# schemas.py
from pydantic import BaseModel, Field
from typing import Literal, List
class Ponto(BaseModel):
t: float
y: float
class DeltaV1(BaseModel):
type: Literal["delta"]
version: Literal[1] = 1
points: List[Ponto] = Field(default_factory=list)# app.py
from schemas import DeltaV1, Ponto
async def publicar_pontos(room_id: str, pts: list[tuple[float, float]]) -> None:
msg = DeltaV1(
type="delta",
points=[Ponto(t=t, y=y) for t, y in pts],
).model_dump()
await broadcast(room_id, msg)Receita 9: Escalar com Redis Pub/Sub (múltiplos workers e várias instâncias)
Conjuntos em memória funcionam apenas dentro de um processo. Em produção, é comum rodar múltiplos workers ou várias instâncias, e cada uma terá suas próprias salas locais. Redis Pub/Sub é um mecanismo de publicação e assinatura onde processos diferentes recebem as mesmas mensagens por um canal. Assim, uma atualização publicada em uma instância alcança conexões mantidas por outras.
O fluxo típico é: a instância que gera eventos publica no Redis, e todas as instâncias assinam o canal e fazem broadcast local para suas próprias conexões. A mensagem precisa conter a sala e o payload para encaminhamento correto. O Redis usado aqui é o cliente assíncrono redis.asyncio, integrando bem com FastAPI. Esse desenho mantém consistência do dashboard mesmo com autoscaling e reinícios.
# redis_bus.py
import json
import redis.asyncio as redis
r = redis.from_url("redis://localhost:6379")
CANAL = "ws:board"
async def publicar(room_id: str, payload: dict) -> None:
await r.publish(CANAL, json.dumps({"room": room_id, "data": payload}, ensure_ascii=False))
async def assinar(callback):
pub = r.pubsub()
await pub.subscribe(CANAL)
async for msg in pub.listen():
if msg.get("type") == "message":
obj = json.loads(msg["data"])
await callback(obj["room"], obj["data"])# app.py
import asyncio
from redis_bus import publicar, assinar
@app.on_event("startup")
async def iniciar_bus():
asyncio.create_task(assinar(lambda room, data: broadcast(room, data)))Receita 10: Deploy sem downtime (drain, reconexão e retomada)
Deploys e reinícios encerram conexões WebSocket, o que pode causar “piscadas” e perda de atualizações. Um padrão simples é enviar um frame de controle informando reinício e fechar a conexão de forma ordenada. Do lado do cliente, a reconexão automática reduz impacto. Para evitar perder dados entre desconectar e reconectar, é útil um cursor, como um timestamp do último evento recebido.
Na retomada, o cliente reconecta com um parâmetro cursor e o servidor pode reenviar deltas perdidos antes de voltar ao fluxo ao vivo. Em cargas maiores, um buffer por sala pode ser mantido em memória ou persistido, dependendo do requisito de consistência. O mais importante é existir uma estratégia explícita, pois quedas e deploys são inevitáveis. Com isso, o “tempo real” se mantém contínuo mesmo com manutenção.
# app.py
from fastapi import FastAPI, WebSocket
from websocket_utils import safe_send_json
app = FastAPI()
async def avisar_reinicio(room_id: str) -> None:
await broadcast(room_id, {"type": "server:restarting"})
# exemplo de endpoint que aceita cursor para retomada simples
@app.websocket("/ws/resume/{room_id}")
async def ws_resume(ws: WebSocket, room_id: str):
cursor = ws.query_params.get("cursor")
await ws.accept()
# Em um cenário real, cursor seria usado para reenviar deltas perdidos
await safe_send_json(ws, {"type": "resume", "room": room_id, "cursor": cursor})Exemplo completo: loop de métricas no servidor e cliente no navegador
Um exemplo prático ajuda a conectar as peças: uma tarefa periódica produz uma métrica e publica para uma sala. O servidor envia mensagens com type bem definido e o cliente interpreta conforme o tipo. O fluxo inclui resposta a ping e reconexão quando o socket fecha. Mesmo sendo simples, esse formato suporta evolução para filas, batching e Redis sem mudar o contrato básico.
No servidor, a métrica pode ser CPU, memória ou contadores internos da aplicação. O loop roda no startup e chama broadcast para a sala correspondente. Em produção, o ideal é tratar erros no loop e evitar que uma exceção encerre a tarefa. No navegador, a conexão lida com mensagens, controle de reinício e fechamento, mantendo o painel ativo.
# app.py
import asyncio
import psutil
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def iniciar_metricas():
async def loop_metricas():
while True:
await asyncio.sleep(0.5)
uso = psutil.cpu_percent()
await broadcast("board:cpu", {"type": "delta", "usage": uso})
asyncio.create_task(loop_metricas())<script>
let ws;
function conectar() {
ws = new WebSocket("ws://localhost:8000/ws/room/board:cpu");
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
if (msg.type === "delta") {
// Exemplo: atualização de um marcador de CPU
// updateGauge(msg.usage);
console.log("CPU:", msg.usage);
}
if (msg.type === "ping") {
ws.send("pong");
}
if (msg.type === "server:restarting") {
setTimeout(conectar, 1000);
}
};
ws.onclose = () => setTimeout(conectar, 1000);
}
conectar();
</script>Diagnóstico de problemas comuns em dashboards WebSocket
Alguns sintomas aparecem com frequência em sistemas de tempo real. CPU alta costuma estar ligada a excesso de mensagens e serialização repetida de JSON, resolvida com batching e payloads menores. Congelamento no cliente costuma envolver ausência de heartbeat ou aba suspensa, e o ping do servidor ajuda a limpar conexões inválidas. Mensagens “sumindo” em parte dos clientes normalmente indicam múltiplos workers sem um barramento compartilhado.
Crescimento de memória geralmente vem de filas sem limite, salas nunca limpas ou buffers acumulando sem publicação. Clientes lentos exigem backpressure explícito, para não travar o fan-out. Mudanças de payload sem versionamento quebram front-ends silenciosamente, o que torna o schema versionado um item estrutural. Com esses pontos cobertos, o comportamento tende a ficar estável e previsível.
Conclusão: disciplina de pequenas decisões para tempo real confiável
Dashboards em tempo real com FastAPI não dependem de truques, mas de um conjunto de padrões consistentes. Autenticação antes de aceitar, isolamento por sala, broadcast seguro e heartbeats reduzem fragilidade e melhoram a experiência. Backpressure com filas e batching evitam que rajadas e clientes lentos provoquem travamentos e consumo excessivo. Para escalar, Redis Pub/Sub conecta processos e instâncias, e uma estratégia de reconexão com cursor reduz impacto de deploys e quedas.
Com contratos de schema e mensagens de controle, o sistema fica fácil de entender e evoluir. O resultado é um fluxo de dados contínuo, com comportamento previsível sob carga e sob falhas reais de rede. A base criada por essas receitas permite crescer do protótipo a um ambiente com múltiplas instâncias sem reinventar o desenho. Esse fechamento estabelece um ciclo completo: conectar, validar, publicar, proteger, escalar e encerrar com segurança.