Sse Python
DodaTech
4 min read
title: "SSE with Python" description: "Learn how to implement Server-Sent Events in Python using Flask, Django, and FastAPI for real-time server-to-client streaming." weight: 16 date: 2026-06-28 lastmod: 2026-06-28 tags: ["apis", "sse"]
Python web frameworks can serve SSE endpoints for real-time streaming. This lesson covers SSE implementations in Flask, Django, and FastAPI with proper streaming response patterns.
## What You'll Learn
- SSE with Flask StreamingResponse
- SSE with Django StreamingHttpResponse
- SSE with FastAPI StreamingResponse
- Generator-based event streaming
- Connection management in Python
## Why It Matters
Python is widely used for data processing and machine learning. SSE enables Python backends to stream results, progress updates, and real-time data to web clients efficiently.
## Real-World Use
A Python-based data processing pipeline uses FastAPI SSE to stream progress updates to a web dashboard. Users see each processing step complete in real time without polling the API.
## Flow Chart
```mermaid
flowchart LR
A[Python Server] --> B[Data Source]
B --> C{SSE Generator}
C --> D[Flask: StreamResponse]
C --> E[Django: StreamHttpResponse]
C --> F[FastAPI: StreamResponse]
D --> G[Client]
E --> G
F --> G
Code Examples
Example 1: Flask SSE with StreamingResponse
from flask import Flask, Response, request
import json
import time
import random
app = Flask(__name__)
@app.route('/events')
def sse_events():
def event_stream():
# Send initial connection event
yield f"event: connected\ndata: {json.dumps({'status': 'connected'})}\n\n"
while True:
# Check if client is still connected
if request.environ.get('wsgi.peer') is None:
break
data = json.dumps({
'time': time.strftime('%H:%M:%S'),
'value': random.randint(1, 100),
})
yield f"data: {data}\n\n"
time.sleep(2)
return Response(
event_stream(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no',
}
)
if __name__ == '__main__':
app.run(threaded=True)
Expected output: Flask SSE endpoint streams random values every 2 seconds to connected clients.
Example 2: FastAPI SSE with StreamingResponse
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
import random
app = FastAPI()
async def event_generator(request: Request):
try:
while True:
# Check if client disconnected
if await request.is_disconnected():
break
data = json.dumps({
'timestamp': str(asyncio.get_event_loop().time()),
'cpu': random.uniform(0, 100),
'memory': random.uniform(0, 100),
})
yield f"data: {data}\n\n"
await asyncio.sleep(1)
except asyncio.CancelledError:
pass
@app.get('/events')
async def sse_endpoint(request: Request):
return StreamingResponse(
event_generator(request),
media_type='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no',
}
)
@app.get('/api/trigger')
async def trigger_event(event: str = 'custom', message: str = ''):
# This would broadcast to all connected clients in a real app
return {'status': 'event would be sent'}
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=8000)
Expected output: FastAPI SSE endpoint streams system metrics using async generator with proper disconnect detection.
Example 3: Django SSE with StreamingHttpResponse
# views.py
import json
import time
from django.http import StreamingHttpResponse
from django.views.decorators.http import require_GET
def sse_stream(request):
def event_stream():
# Check client connection
yield f"event: connected\ndata: {json.dumps({'status': 'streaming'})}\n\n"
events_sent = 0
while events_sent < 100: # Max 100 events
data = json.dumps({
'event_id': events_sent,
'message': f'Event number {events_sent}',
'timestamp': time.strftime('%Y-%m-%dT%H:%M:%S'),
})
yield f"data: {data}\n\n"
events_sent += 1
time.sleep(1)
response = StreamingHttpResponse(
event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
response['Connection'] = 'keep-alive'
return response
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('events/', views.sse_stream, name='sse-stream'),
]
# settings.py - Disable GZip for SSE
GZIP_CONTENT_TYPES = (
'text/css',
'text/javascript',
'application/javascript',
# Note: 'text/event-stream' is NOT included
)
Expected output: Django SSE endpoint streams 100 events at 1-second intervals with proper headers.
Common Mistakes
| Mistake | Explanation |
|---|---|
| Not disabling GZip compression | GZip buffers SSE streams; ensure text/event-stream is not compressed |
| Using WSGI with blocking operations | WSGI (Flask, Django) blocks on time.sleep; use async frameworks for long-lived streams |
| Forgetting X-Accel-Buffering header | NGINX behind the scenes may buffer SSE; set X-Accel-Buffering: no |
| Not checking client disconnection | Without disconnect detection, generators run indefinitely, leaking resources |
| Using threading for every connection | Thread-based servers may not scale to thousands of SSE connections |
Practice Questions
- How do you implement SSE in Flask?
- How does FastAPI's async SSE differ from Flask's synchronous SSE?
- How do you detect client disconnection in Python SSE?
- What header is needed to disable NGINX buffering for SSE?
- How do you broadcast events to all connected clients in Python?
Challenge
Build a Python SSE server that streams real-time stock prices from a data source. Implement per-client topic subscriptions so clients can subscribe to specific stock symbols and receive only relevant price updates.
FAQ
Mini Project
Build a Python-based real-time data pipeline monitor with FastAPI SSE. The server monitors a data processing pipeline and streams metrics (records processed, errors, throughput) to a web dashboard. Include per-pipeline subscription filtering.
What's Next
Learn about advanced SSE patterns in Node.js
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro