Building SSE in Django with Django Channels

Intro to SSE

Server-Sent Events (SSE) let servers push updates to the browser over plain HTTP. An SSE client opens a single long-lived HTTP connection and the server streams text events down that connection as they become available.

Difference from normal request–response:

• Normal: client requests → server responds once.

• SSE: client opens connection → server keeps it open → server sends events as needed.

Think of SSE as a one-way WebSocket (server → client only). It’s a great fit for notifications, progress updates, live feeds, and other server-to-client streams where you don’t need client → server real-time channels.

When to prefer SSE over WebSockets:

• You only need server → client updates.

• You prefer a simpler HTTP-based approach (no websocket handshake).

• You want built-in browser features like automatic reconnection, event naming, and last-event-id support.

---

Stack

• Django — main framework

• Django Channels — async primitives and channel layer/groups

• Daphne — ASGI server

• Redis — channel layer backend (message broker)

---

Note: This demo is currently being used in internal and UAT environment, the actual production environment behaviour and use-cases are yet to be seen.

Install

pip install channels daphne channels-redis

---

Basic settings

Add Daphne and Channels to INSTALLED_APPS and configure the channel layer (example using Redis):

# settings.py
INSTALLED_APPS = [
    # ...
    "daphne",
    "channels",
    # ...
]

ASGI_APPLICATION = "myproject.asgi.application"

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
            "capacity": 2000,
            "expiry": 60,
            "group_expiry": 3600,
        },
    },
}

---

ASGI routing

Route the SSE path to an AsyncHttpConsumer. Put specific Channels routes before the fallback get_asgi_application() so normal Django views still work.

# myproject/asgi.py
import os
import django
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
django.setup()

from myapp.consumers import SSEConsumer  # your SSE consumer

application = ProtocolTypeRouter(
    {
        "http": URLRouter(
            [
                path("sse/", SSEConsumer.as_asgi()),  # SSE route first
                path("", get_asgi_application()),     # fallback to Django
            ]
        ),
        # "websocket": ... if you also use websockets
    }
)

Order matters: route the SSE URL first so Channels picks the consumer for that path.

---

Base SSE Consumer

This consumer:

• Joins a channel-layer group so other parts of the app can broadcast to connected clients.

• Sets SSE headers and streams events.

• Sends periodic heartbeats to keep proxies from closing the connection.

• Handles events delivered via the channel layer.

• For More Check Here: Channels Docs

# myapp/consumers.py
import asyncio
import json
from channels.generic.http import AsyncHttpConsumer

class BaseSSEConsumer(AsyncHttpConsumer):
    group_name = None
    keep_running = True

    async def handle(self, body):
        self.group_name = await self.get_group_name()
        await self.channel_layer.group_add(self.group_name, self.channel_name)

        await self.send_headers(
            headers=[
                (b"Cache-Control", b"no-cache"),
                (b"Content-Type", b"text/event-stream"),
                (b"Transfer-Encoding", b"chunked"),
                (b"Connection", b"keep-alive"),
            ]
        )

        self.heartbeat_task = asyncio.create_task(self._heartbeat())

    async def _heartbeat(self):
        while self.keep_running:
            try:
                await self.send_body(b"event: ping\ndata: pong\n\n", more_body=True)
            except Exception:
                break
            await asyncio.sleep(30)

    async def sse_event(self, event: dict):
        event_key = event.get("event")
        data_payload = event.get("data")
        data = f"event: {event_key}\ndata: {json.dumps(data_payload)}\n\n"
        await self.send_body(data.encode("utf-8"), more_body=True)

    async def http_request(self, message):
        await self.handle(b"")

    async def http_disconnect(self, message):
        self.keep_running = False
        if hasattr(self, "heartbeat_task"):
            self.heartbeat_task.cancel()
        if self.group_name:
            await self.channel_layer.group_discard(self.group_name, self.channel_name)

(Adapt get_group_name, authentication, and event dispatching to your app's needs.)

---

Broadcasting events from Django Use Channels’ channel layer to send events to the SSE group.

Synchronous view/function:

# myapp/utils.py
# any module where you want to broadcast
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

def broadcast_notification(data):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        "notifications",
        {
            "type": "sse.event",
            "event": "notification",
            "data": data,
        },
    )

# Using it in a view.
# myapp/views.py
def mark_done(request):
    # ... do work
    broadcast_notification({"msg": "Task completed", "id": 123})
    return JsonResponse({"ok": True})

Nginx configuration for SSE

For SSE to work reliably behind Nginx you must disable buffering and ensure long timeouts. Add an upstream block for Daphne (TCP or unix socket) and a location for the SSE endpoint.

Example upstream (TCP):

upstream sse_upstream {
    server 127.0.0.1:9095;
}

Insert the provided SSE location block into your server block so Nginx streams responses immediately:

location /sse/ {
    proxy_pass http://sse_upstream$request_uri;

    proxy_http_version 1.1;

    proxy_set_header Host $host;
    proxy_set_header Connection '';
    proxy_set_header X-Accel-Buffering no;

    proxy_buffering off;
    proxy_cache off;

    chunked_transfer_encoding on;
    gzip off;

    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;

    add_header Cache-Control no-cache;
}

Notes:

• proxy_buffering off and X-Accel-Buffering no help Nginx stream events immediately to clients.

• proxy_http_version 1.1 and clearing Connection header let chunked responses flow correctly.

• Long proxy_read_timeout/proxy_send_timeout keep long-lived SSE connections from being closed prematurely.

• chunked_transfer_encoding on and gzip off avoid interfering with streaming behavior.

---

Running Daphne

You can run Daphne directly for testing or under a process manager in production. Here is the exact command you provided (keeps HTTP timeout disabled and exposes Daphne on 0.0.0.0:9095):

uv run daphne \
  -b 0.0.0.0 \
  -p 9095 \
  --http-timeout 0 \
  --proxy-headers \
  --access-log - \
  "myproject.asgi:application"

Thank you.