Django 3.1: StreamingHttpResponse with an async generator

Question

Documentation for Django 3.1 says this about async views:

The main benefits are the ability to service hundreds of connections without using Python threads. This allows you to use slow streaming, long-polling, and other exciting response types.

I believe that "slow streaming" means we could implement an SSE view without monopolizing a thread per client, so I tried to sketch a simple view, like so:

async def stream(request):

    async def event_stream():
        while True:
            yield 'data: The server time is: %s\n\n' % datetime.datetime.now()
            await asyncio.sleep(1)

    return StreamingHttpResponse(event_stream(), content_type='text/event-stream')

(note: I adapted the code from this response)

Unfortunately, when this view is invoked, it raises the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/asgiref/sync.py", line 330, in thread_handler
    raise exc_info[1]
  File "/usr/local/lib/python3.7/site-packages/django/core/handlers/exception.py", line 38, in inner
    response = await get_response(request)
  File "/usr/local/lib/python3.7/site-packages/django/core/handlers/base.py", line 231, in _get_response_async
    response = await wrapped_callback(request, *callback_args, **callback_kwargs)
  File "./chat/views.py", line 144, in watch
    return StreamingHttpResponse(event_stream(), content_type='text/event-stream')
  File "/usr/local/lib/python3.7/site-packages/django/http/response.py", line 367, in __init__
    self.streaming_content = streaming_content
  File "/usr/local/lib/python3.7/site-packages/django/http/response.py", line 382, in streaming_content
    self._set_streaming_content(value)
  File "/usr/local/lib/python3.7/site-packages/django/http/response.py", line 386, in _set_streaming_content
    self._iterator = iter(value)
TypeError: 'async_generator' object is not iterable

To me, this shows that StreamingHttpResponse doesn't currently support async generators.

I tried to modify StreamingHttpResponse to use async for but I wasn't able to do much.

Any idea how I could do that?

Answer 1

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.