SSE with Django — Complete Guide
In this tutorial, you will learn about SSE with Django. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement Server-Sent Events in Django using StreamingHttpResponse, manage SSE views, handle client disconnection, broadcast events, and integrate real-time updates into Django applications.
What You Learn
You will learn how to create SSE endpoints in Django using StreamingHttpResponse, handle client disconnection, broadcast events to multiple clients, use Django channels for scalable SSE, and integrate SSE with Django REST Framework.
Why It Matters
Django traditionally uses request-response cycles. SSE breaks this pattern by keeping a connection open for streaming. Integrating SSE into Django enables real-time features without switching to a different framework.
Real-World Use
DodaTech's Django-based admin panel uses SSE for live monitoring. When an admin views the dashboard, Django streams CPU metrics, request counts, and error rates. Multiple admins see the same data simultaneously.
Basic SSE View
# views.py
import json
import time
from django.http import StreamingHttpResponse
from django.views import View
class SSEView(View):
def get(self, request, *args, **kwargs):
def event_stream():
counter = 0
while True:
counter += 1
data = json.dumps({
'count': counter,
'time': time.time(),
})
yield f"id: {counter}\ndata: {data}\n\n"
time.sleep(1)
if counter >= 10:
break
response = StreamingHttpResponse(
streaming_content=event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
response['X-Accel-Buffering'] = 'no'
return response
# urls.py
from django.urls import path
from .views import SSEView
urlpatterns = [
path('events/', SSEView.as_view(), name='sse-events'),
]
Client Disconnection Handling
# views.py
import json
import time
from django.http import StreamingHttpResponse
def sse_stream(request):
def event_stream():
heartbeat_count = 0
connected = True
def check_connection():
nonlocal connected
if request.environ.get('wsgi.websocket'):
return
# Check if client is still connected
try:
if not request.META.get('HTTP_ACCEPT'):
connected = False
except Exception:
connected = False
try:
# Send initial event
yield f"event: connected\ndata: {json.dumps({'status': 'streaming'})}\n\n"
while connected:
# Send heartbeat every 15 seconds
heartbeat_count += 1
if heartbeat_count % 15 == 0:
yield f": heartbeat {time.time()}\n\n"
# Check if client disconnected
check_connection()
data = json.dumps({
'message': 'update',
'timestamp': time.time(),
})
yield f"event: update\ndata: {data}\n\n"
time.sleep(1)
except GeneratorExit:
# Client disconnected
print("SSE client disconnected")
except Exception as e:
print(f"SSE error: {e}")
response = StreamingHttpResponse(
streaming_content=event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
return response
Broadcasting with Redis
# sse_broadcast.py
import json
import threading
import redis
from django.http import StreamingHttpResponse
r = redis.Redis()
class SSEClientManager:
_instance = None
_lock = threading.Lock()
def __new__(cls):
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.clients = set()
cls._instance.client_lock = threading.Lock()
return cls._instance
def add_client(self, queue):
with self.client_lock:
self.clients.add(queue)
def remove_client(self, queue):
with self.client_lock:
self.clients.discard(queue)
def broadcast(self, event, data, exclude=None):
message = json.dumps({'event': event, 'data': data})
r.publish('sse_broadcast', message)
def get_client_count(self):
with self.client_lock:
return len(self.clients)
manager = SSEClientManager()
def sse_broadcast_view(request):
def event_stream():
pubsub = r.pubsub()
pubsub.subscribe('sse_broadcast')
manager.add_client(pubsub)
try:
yield f"event: connected\ndata: {json.dumps({'clients': manager.get_client_count()})}\n\n"
for message in pubsub.listen():
if message['type'] == 'message':
payload = json.loads(message['data'])
yield f"event: {payload['event']}\ndata: {json.dumps(payload['data'])}\n\n"
except GeneratorExit:
pass
finally:
pubsub.unsubscribe()
manager.remove_client(pubsub)
response = StreamingHttpResponse(
streaming_content=event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
return response
def publish_event(request):
if request.method == 'POST':
import json as json_module
data = json_module.loads(request.body)
manager.broadcast(data.get('event', 'message'), data.get('data', {}))
return JsonResponse({'sent': True, 'clients': manager.get_client_count()})
return JsonResponse({'error': 'POST required'}, status=405)
SSE with Django Channels
# consumers.py
import json
from channels.generic.http import AsyncHttpConsumer
class SSEConsumer(AsyncHttpConsumer):
async def handle(self, body):
await self.send_headers(headers=[
(b'Content-Type', b'text/event-stream'),
(b'Cache-Control', b'no-cache'),
(b'X-Accel-Buffering', b'no'),
])
await self.send_body(
f"event: connected\ndata: {json.dumps({'status': 'ready'})}\n\n".encode(),
more_body=True,
)
self.keep_alive = True
self.counter = 0
while self.keep_alive:
self.counter += 1
data = json.dumps({
'count': self.counter,
'time': __import__('time').time(),
})
await self.send_body(
f"id: {self.counter}\ndata: {data}\n\n".encode(),
more_body=True,
)
await asyncio.sleep(1)
if self.counter >= 20:
break
async def disconnect(self):
self.keep_alive = False
# routing.py
from django.urls import re_path
from .consumers import SSEConsumer
websocket_urlpatterns = [
re_path(r'^sse/stream/$', SSEConsumer.as_asgi()),
]
SSE with Django REST Framework
# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from django.http import StreamingHttpResponse
import json
import time
class SSEAPIView(APIView):
def get(self, request, *args, **kwargs):
def event_stream():
channel = request.query_params.get('channel', 'default')
yield f"event: connected\ndata: {json.dumps({'channel': channel})}\n\n"
for i in range(20):
data = {
'channel': channel,
'index': i,
'timestamp': time.time(),
}
yield f"event: {channel}_update\ndata: {json.dumps(data)}\n\n"
time.sleep(1)
response = StreamingHttpResponse(
streaming_content=event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
return response
# urls.py
from django.urls import path
from .views import SSEAPIView
urlpatterns = [
path('api/sse/', SSEAPIView.as_view(), name='api-sse'),
]
Common Mistakes
1. Response Buffering
Django's GZipMiddleware or other middleware buffers the response. Disable compression for SSE views. Use @gzip_page or conditionally disable.
2. Blocking the Event Loop
Django's synchronous views block the entire worker. For multiple SSE clients, use a separate worker or async views (Django 3.1+).
3. No Keep-Alive
Proxies (nginx, Apache) close idle connections without periodic data. Send heartbeat comments every 15-30 seconds.
4. GeneratorExit Not Handled
When the client disconnects, Django raises GeneratorExit in the generator. Catch it to clean up resources properly.
5. Using TemplateResponse
SSE views must return StreamingHttpResponse, not TemplateResponse or regular HttpResponse.
Practice Questions
1. What Django class is used for SSE responses?
StreamingHttpResponse. It streams content as a generator, keeping the connection open for multiple events.
2. How do you handle client disconnection in Django SSE?
Catch GeneratorExit in the generator function. This exception is raised when the client disconnects.
3. How do you broadcast events to multiple Django SSE clients?
Use Redis pub/sub. Each SSE client subscribes to a Redis channel. Publishers send messages, and the pub/sub distributes to all subscribers.
4. What middleware can break SSE in Django?
GZipMiddleware compresses the response buffer, delaying delivery. Ensure compression is disabled for SSE views.
Challenge
Build a Django SSE system for a monitoring dashboard: SSE view streaming CPU/memory metrics, Redis pub/sub for broadcasting to all dashboard clients, POST endpoint to inject test metrics, client count tracking, and heartbeat keep-alive.
FAQ
Mini Project: Django SSE Monitor
# views.py
import json
import time
import random
from django.http import StreamingHttpResponse, JsonResponse
from django.views.decorators.http import require_POST
def monitor_stream(request):
def event_stream():
yield f"event: connected\ndata: {json.dumps({'status': 'monitoring'})}\n\n"
try:
while True:
metrics = {
'cpu': random.uniform(0, 100),
'memory': random.uniform(0, 100),
'disk': random.uniform(0, 100),
'requests': random.randint(0, 1000),
'time': time.time(),
}
yield f"event: metrics\ndata: {json.dumps(metrics)}\n\n"
yield f": heartbeat {time.time()}\n\n"
time.sleep(2)
except GeneratorExit:
pass
response = StreamingHttpResponse(
streaming_content=event_stream(),
content_type='text/event-stream',
)
response['Cache-Control'] = 'no-cache'
response['X-Accel-Buffering'] = 'no'
return response
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('sse/monitor/', views.monitor_stream, name='monitor-stream'),
]
What's Next
Now that you understand SSE with Django, learn SSE with FastAPI, then explore SSE with plain Node.js.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro