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 APIRouter, BackgroundTasks, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
 

Building a signup endpoint in FastAPI

dependency-injection request-validation background-tasks
Intermediate 8 steps
javascript 
const ROLE_PERMISSIONS = {
  admin: ['users:read', 'users:write', 'billing:read', 'billing:write'],
  manager: ['users:read', 'billing:read'],
  member: ['users:read'],

Role-based permissions middleware in Express

authorization middleware rbac
Intermediate 9 steps
python 
from django import forms
from django.utils import timezone
 
from .models import Reservation

Multi-field validation in a Django ModelForm

form validation cross-field validation modelform
Intermediate 7 steps
python 
from django import template
from django.urls import reverse, NoReverseMatch
from django.utils.html import format_html
 

Active nav-link template tags in Django

template tags url routing active state
Intermediate 7 steps
javascript 
import { useReducer, useEffect } from "react";
 
const initialState = { status: "idle", data: null, error: null };
 

Building a data-fetching hook in React

custom-hooks usereducer data-fetching
Intermediate 9 steps
php 
<?php
 
namespace App\Broadcasting;
 

Authorizing presence channels in Laravel

broadcasting authorization presence-channels
Intermediate 3 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