Most metrics start life as a file. You have counters in a database, or a JSONL file some job appends to, and to see what changed you re-run a script and read the tail. That works until you want to watch it. Then you start refreshing, and refreshing is just polling done by a human.
Here's the shape of the whole thing. One producer, many browsers, and one bounded queue per browser.
The complete runnable code, including the test suite, is in this gist.
Why not WebSockets?
A dashboard never talks back, so a WebSocket buys you a bidirectional channel you never use. You pay for it with a protocol upgrade, a separate deployment story, and reconnection logic you write yourself. SSE is plain HTTP. It passes through proxies, it works with the auth middleware you already have, and reconnection is a browser feature rather than your code.
Reach for WebSockets when the client genuinely sends messages: chat, collaborative editing, multiplayer. For numbers on a screen, SSE is less machinery for the same result.
Fan out with one bounded queue per client
Give every connected browser its own bounded asyncio.Queue and drop the oldest message when it fills, so one slow client can never stall the producer. The load-bearing word is bounded.
import asyncio, json, random
from contextlib import asynccontextmanager, suppress
from typing import AsyncIterator
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
class Broker:
"""Pub/sub fan-out. Every subscriber owns a BOUNDED queue, so a slow
browser tab can never apply backpressure to the producer."""
def __init__(self, maxsize: int = 10) -> None:
self._maxsize = maxsize
self._subscribers: set[asyncio.Queue[str]] = set()
self.dropped = 0
@property
def subscriber_count(self) -> int:
return len(self._subscribers)
def publish(self, message: str) -> None:
"""Never awaits. If a subscriber is full, evict its oldest message
(stale metrics are worthless) and count the drop."""
for queue in self._subscribers:
try:
queue.put_nowait(message)
except asyncio.QueueFull:
queue.get_nowait()
queue.put_nowait(message)
self.dropped += 1
@asynccontextmanager
async def subscribe(self) -> AsyncIterator[asyncio.Queue[str]]:
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=self._maxsize)
self._subscribers.add(queue)
try:
yield queue
finally:
self._subscribers.discard(queue) # a disconnect always cleans up
broker = Broker()
publish() never awaits, and that's the whole design. If you await a slow client, one laptop on hotel wifi backs up the producer and stalls every other viewer. Bounding the queue and dropping the oldest entry turns that outage into a counter. For metrics, that trade is obviously right, because nobody wants a stale number delivered late.
Staying synchronous buys a second property for free. Because publish() never yields control, no client can disconnect mid-broadcast and mutate the set underneath the loop.
subscribe() is an async context manager, so a dropped connection always removes its queue. That matters more than it sounds. A leak here stays invisible until you've run for a week and every disconnected tab is still holding memory.
Start the producer with lifespan, not on_event
Start background producers in FastAPI's lifespan context manager rather than @app.on_event("startup"). If you pass lifespan, the old startup and shutdown handlers never run at all. FastAPI's own docs are blunt here. "If you provide a lifespan parameter," they warn, "startup and shutdown event handlers will no longer be called."
async def metrics_producer(target: Broker) -> None:
tick = 0
while True:
tick += 1
target.publish(json.dumps({
"tick": tick,
"rps": round(random.uniform(80, 240), 1),
"latency_ms": round(random.uniform(12, 90), 1),
"clients": target.subscriber_count,
}))
await asyncio.sleep(1.0)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
task = asyncio.create_task(metrics_producer(broker))
try:
yield
finally:
task.cancel()
with suppress(asyncio.CancelledError):
await task
app = FastAPI(lifespan=lifespan)
Swap the random numbers for your real stats source. The cancel-and-suppress in the finally block is what makes shutdown quiet instead of dumping a cancellation traceback.
The endpoint, and what a frame actually looks like
An SSE frame is just text with a blank line at the end.
async def event_stream(request: Request) -> AsyncIterator[str]:
async with broker.subscribe() as queue:
event_id = 0
while True:
try:
payload = await asyncio.wait_for(queue.get(), 15.0)
except TimeoutError:
yield ": ping\n\n" # comment frame: keeps idle proxies from reaping us
continue
if await request.is_disconnected():
break # drops this one message; fine for metrics, not for orders
event_id += 1
yield f"id: {event_id}\nevent: metrics\ndata: {payload}\n\n"
@app.get("/events")
async def events(request: Request) -> StreamingResponse:
return StreamingResponse(
event_stream(request),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # nginx would otherwise buffer the stream
},
)
Three details carry the weight. A blank line terminates a frame. A line starting with : is a comment, which is why the heartbeat is spelled : ping. And data: has to be a single line, which json.dumps() guarantees for free.
The timeout doubles as the heartbeat. A plain await queue.get() gives you nowhere to emit a keepalive on an idle stream, so an idle connection sits silent until some proxy decides it's dead. Wrapping the get in wait_for lets one loop handle both data and liveness.
Here's what curl sees, at roughly one frame per second:
id: 1
event: metrics
data: {"tick": 18, "rps": 121.9, "latency_ms": 40.3, "clients": 1}
id: 2
event: metrics
data: {"tick": 19, "rps": 205.5, "latency_ms": 62.6, "clients": 1}
The browser needs nine lines
The browser side is nine lines: new EventSource("/events"), one event listener, and no reconnect logic at all.
// EventSource reconnects on its own and sends the last `id:` back as a
// Last-Event-ID header, so the server *can* resume you, if it keeps a replay
// buffer. This one doesn't. It just picks up live.
const es = new EventSource("/events");
es.addEventListener("metrics", (e) => {
document.getElementById("state").textContent = "live";
for (const [k, v] of Object.entries(JSON.parse(e.data))) {
const el = document.getElementById(k);
if (el) el.textContent = v;
}
});
es.onerror = () => { document.getElementById("state").textContent = "reconnecting..."; };
No reconnect loop, no exponential backoff, no library. Kill the server and the numbers freeze while the label flips to reconnecting.... Start it again and they resume.
The four things that silently break (hanging tests, dead producers, nginx, and the six-connection cap)
Each of these looks like a bug in your own code. None of them are.
Your tests hang forever. httpx.ASGITransport buffers the entire response body before returning, so client.stream("GET", "/events") against an endless stream never returns at all. Even response.status_code is unreachable. It looks exactly like a deadlock in your own broker. Drive the ASGI app directly and cancel after N frames, or run a real server in a fixture.
Your producer never starts under test. ASGITransport doesn't run lifespan, so nothing publishes and every test reads zero frames. TestClient does run it, but only as a context manager. A bare TestClient(app) without with skips lifespan too, and just as silently. If you export your lifespan function you can enter it in a fixture with async with lifespan(app): and skip the extra dependency. Migrating from on_event to lifespan changes test behaviour without a warning, which is the real trap.
nginx buffers your stream. Every local curl looks perfect without X-Accel-Buffering: no, and then production delivers your "real-time" events in batches every few KB. This one stays invisible until you deploy behind a proxy.
You get six connections per domain. Over HTTP/1.1 most browsers cap simultaneous connections to one domain at six, and that budget is shared across tabs. A few open dashboards can starve the rest of your app. Over HTTP/2 the negotiated stream limit defaults to around 100, so serving the thing over HTTP/2 makes the problem disappear.
Bound the queue, drop the oldest
The streaming is never the hard part. Roughly 120 lines gets you a live dashboard, and most of it is ordinary FastAPI. The hard part is deciding what happens when one client can't keep up. Bound the queue, drop the oldest frame, count the drops. A number that arrives late is worse than a number that never arrives, because you'll trust it.
Clone the gist, run uvicorn app:app, and open two tabs to watch the client counter move. Then shorten the heartbeat interval and watch the bytes, rather than reading the code and assuming.
I write these from real work at astraedus.dev, where I build apps and tools. Building something, or stuck on something like this? Reach me there or at [email protected].