python 47 lines · 8 steps

Streaming export progress with SSE in FastAPI

A FastAPI route pushes live job progress to the client over Server-Sent Events using an async generator.

Explained by highlit
FastAPI Advanced server-sent-events async-generators streaming authorization cancellation
1import asyncio
2import json
3 
4from fastapi import APIRouter, Depends, HTTPException
5from sse_starlette.sse import EventSourceResponse
6 
7from .deps import get_current_user
8from .jobs import ExportJob, job_registry
9 
10router = APIRouter(prefix="/exports", tags=["exports"])
11 
12 
13@router.get("/{job_id}/stream")
14async def stream_export_progress(job_id: str, user=Depends(get_current_user)):
15 job = job_registry.get(job_id)
16 if job is None or job.owner_id != user.id:
17 raise HTTPException(status_code=404, detail="Export job not found")
18 
19 async def event_publisher():
20 last_percent = -1
21 try:
22 while not job.is_finished:
23 await job.wait_for_update(timeout=15)
24 
25 if job.percent != last_percent:
26 last_percent = job.percent
27 yield {
28 "event": "progress",
29 "id": str(job.revision),
30 "data": json.dumps(
31 {"percent": job.percent, "stage": job.stage}
32 ),
33 }
34 else:
35 yield {"event": "ping", "data": ""}
36 
37 yield {
38 "event": "complete" if job.succeeded else "error",
39 "data": json.dumps(
40 {"download_url": job.download_url, "error": job.error}
41 ),
42 }
43 except asyncio.CancelledError:
44 job.detach_listener()
45 raise
46 
47 return EventSourceResponse(event_publisher())
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Server-Sent Events let a server push incremental updates to a client over one long-lived HTTP response.
  2. 2An async generator is a natural fit for streaming because each yield emits one event without buffering everything.
  3. 3Handling CancelledError lets you clean up listeners when a client disconnects mid-stream.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
go 
package middleware
 
import (
	"net/http"

Per-plan export limits in Gin middleware

middleware rate-limiting authorization
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Streaming export progress with SSE in FastAPI — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code