[Fixed]-Django 3.1: StreamingHttpResponse with an async generator

Honestly it is not supported natively by Django, but I have a solution for you using Daphne (which is also using in Django channels).

Created own StreamingHttpResponse class that is able to retrieve data stream from async methods and provide it to synchronous part of Django.

import asyncio

# By design asyncio does not allow its event loop to be nested.
# Trying to do so will give the error "RuntimeError: This event loop is already running".
# This library solves that problem.
import nest_asyncio

from django.http.response import StreamingHttpResponse


class AsyncStreamingHttpResponse(StreamingHttpResponse):

    def __init__(self, streaming_content=(), *args, **kwargs):
        sync_streaming_content = self.get_sync_iterator(streaming_content)
        super().__init__(streaming_content=sync_streaming_content, *args, **kwargs)

    @staticmethod
    async def convert_async_iterable(stream):
        """Accepts async_generator and async_iterator"""
        return iter([chunk async for chunk in stream])

    def get_sync_iterator(self, async_iterable):
        nest_asyncio.apply()

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(self.convert_async_iterable(async_iterable))
        return result

Also, you’ll need to run your Django web-server using Daphne to support Server Sent Events (SSE) properly. It is officially support by "Django Software Foundation" and has similar syntax to gunicorn, but uses asgi.py instead of wsgi.py.

To use it – you can install using: pip install daphne

And change command from: python manage.py runserver
to something like: daphne -b 0.0.0.0 -p 8000 sse_demo.asgi:application.

Not sure if it will work with gunicorn.

Let me know if you’ll have any more questions.

Another way to do SSE is to use special library django-eventstream:

Add following to HTML page that will consume data:

<script src="{% static 'django_eventstream/eventsource.min.js' %}"></script>
<script src="{% static 'django_eventstream/reconnecting-eventsource.js' %}"></script>

var es = new ReconnectingEventSource('/events/');

es.addEventListener('message', function (e) {
    console.log(e.data);
}, false);

es.addEventListener('stream-reset', function (e) {
    // ... client fell behind, reinitialize ...
}, false);

For backend you’ll need to properly setup Django, and later you’ll be able to call following method in any view/task/signal/method that needs to do Server Side Event (SSE):

Add following view that will produce data (events):

# from django_eventstream import send_event

send_event('test', 'message', {'text': 'hello world'})

I created a decorator named stream that can be used with a coroutine function to make it compatible with Django’s StreamingHttpResponse. Here’s an example:

import asyncio
import functools

from django.http import StreamingHttpResponse


def stream(coroutine_function):
    @functools.wraps(coroutine_function)
    def wrapper(*args, **kwargs):
        coroutine = coroutine_function(*args, **kwargs)
        try:
            while True:
                yield asyncio.run(coroutine.__anext__())
        except StopAsyncIteration:
            pass
    return wrapper


@stream
async def chunks():
    for char in 'Hello, world!':
        yield char
        await asyncio.sleep(1)


async def index(request):
    return StreamingHttpResponse(chunks())

I also needed to add nest_asyncio and call apply() at the top of the settings.py file like:

import nest_asyncio
nest_asyncio.apply()

The nest_asyncio dependency supports calling asyncio.run from the wrapper function created by the stream decorator.

Finally, Django’s asgi can be run using uvicorn through gunicorn like:

$ gunicorn -k uvicorn.workers.UvicornWorker www.asgi:application

It seems you have to use something like django-channel :

Channels augments Django to bring WebSocket, long-poll HTTP, task
offloading and other async support to your code, using familiar Django
design patterns and a flexible underlying framework that lets you not
only customize behaviours but also write support for your own
protocols and needs.