Scenario:

I recently encountered a situation in my personal project where I needed to create a feature similar to the "latest" or "new news" button, like the ones you see on Google News or Inshorts.

What seems upfront?

A simple solution could be to create an endpoint for long polling after a short delay, like 5 or 10 seconds.
- This idea was rejected because it's not scalable.
Web sockets are too complicated for this because they are hard to set up, and our app doesn't send any updates or acknowledgments to our backend system.

What we chose and why?

SSE - Server side events is a good contender for one way push from backend to frontend.

Flow -

flow diagram image

Implementation

Django - The web framework for perfectionists with deadlines

Daphne - is a pure-Python ASGI server for UNIX, maintained by members of the Django project. It serves as the reference server for ASGI.

Steps

add daphne in requirements.txt file and run command
1. make sure to install pip before this.
  1. ```
    pip3 install -r requirements.txt
```
2. Now add daphne in installed apps.
  1. ```
    INSTALLED_APPS = [
        "daphne",
        ...,
    ]
```
3. lets add an endpoint in urls.py to expose the sse.
  1. ```
    path("sse/", FeedSSEView.as_view()),
```
4. lets create the view in views.py to handle the SSE implementation

We will be using redis pubSub and will use it to subscribe to a channel for Instant Updates and Decoupled Architecture.

from asgiref.sync import sync_to_async
from django.http import StreamingHttpResponse

class SSEView(APIView):
    authentication_classes = []
    permission_classes = []
    renderer_classes = [ServerSentEventRenderer]

    def get(self, request, *args, **kwargs):
        async def event_stream():
            try:
                logger.info("initialising pubsub--")
                pubsub = RedisClient().redis_client.pubsub()
                logger.info("Init done pubsub--")
                pubsub.subscribe("some_channel")
                logger.info("Subscribed success to some_channel --")
                while True:
                    message = await sync_to_async(pubsub.get_message)(
                        ignore_subscribe_messages=True, timeout=1.0
                    )
                    if message:
                        logger.info(f"received finfeed_channel msg {message}")
                        yield f"event: {'bulk_update'}\ndata: {message['data']}\n\n"
                    await asyncio.sleep(0.01)  # Prevent tight loop
            except Exception as e:
                logger.exception(f"error in event_stream {e}")
                pubsub.close() # close finally
                raise e

        try:
            return StreamingHttpResponse(
                event_stream(),
                content_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "X-Accel-Buffering": "no",
                    "Access-Control-Allow-Origin": "*",
                },
            )
        except Exception as e:
            logger.exception(f"error in SSEView {e}")
            return Response(
                {"message": "Something went wrong. Please try again later!"},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )

Our SSE renderer would be used for rendering data as Server-Sent Events.

  from rest_framework.renderers import BaseRenderer

  class ServerSentEventRenderer(BaseRenderer):
      media_type = "text/event-stream"
      format = "sse"

      def render(self, data, accepted_media_type=None, renderer_context=None):
          logger.info(f"sse render data is {data}--")
          try:
              sse_data = (
                  f"event: {event['event']}\n" f"data: {event['data']}\n\n"
                  for event in data
                  if isinstance(event, dict) and "event" in event and "data" in event
              )
              sse_data = "".join(sse_data)
              return sse_data.encode()
          except TypeError as e:
              logger.exception(f"SSE type render error: {e}")
              return b"Error rendering SSE data"
          except Exception as e:
              logger.exception(f"SSE render error: {e}")
              return b"Error rendering SSE data"

Lets run the django server via cmd -

```
  python3 manage.py runserver
```

You would see daphne related logs as

  INFO 2025-06-23 06:21:38,539 server 47582 6146600960 HTTP/2 support not enabled (install the http2 and tls Twisted extras)
  INFO 2025-06-23 06:21:38,540 server 47582 6146600960 Configuring endpoint tcp:port=8000:interface=127.0.0.1
  INFO 2025-06-23 06:21:38,540 server 47582 6146600960 Listening on TCP address 127.0.0.1:8000

Lets test

  curl -N -H "Accept: text/event-stream" -H "Authorization: Bearer <jwtToken>" 127.0.0.1:8000/api/sse/

when a new event is generated - you would get msg as.

  event: {'bulk_update'}
  data: {<response_json>}

Deployment

We will use Uvicorn for handling the SSE endpoint and for rest the Gunicorn with nginx as web server.

difference?

Gunicorn: Traditional, synchronous, process manager for WSGI apps.

Uvicorn: Modern, asynchronous, speed-optimized for ASGI apps.

You could make your uvicorn.sh as

currently it uses port 8001 for serving uvicorn requests, we can modify it to work on socket.

[Unit]
Description=Uvicorn ASGI Server
After=network.target

[Service]
User=[USER]
Group=[GROUP]
WorkingDirectory=/home/[USER]/[PROJECT_MAIN_DIR_NAME]
ExecStart=/home/[USER]/venv/bin/uvicorn test.asgi:application --host 0.0.0.0 --port 8001
Restart=always

[Install]
WantedBy=multi-user.target

Daphne for SSE in Django

Table of contents