python - Django 3.1 : StreamingHttpResponse with an async generator

Question

Documentation for Django 3.1关于异步 View 是这样说的:

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.

我相信“慢流”意味着我们可以实现 SSE View 而不为每个客户端独占一个线程，所以我尝试绘制一个简单的 View ，如下所示:

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')

(注意:我改编了来自 this response 的代码)
不幸的是，当调用此 View 时，会引发以下异常:

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

对我来说，这表明 StreamingHttpResponse目前不支持异步生成器。
我试图修改 StreamingHttpResponse使用 async for但我无能为力。
知道我怎么做吗？

Answer 1

老实说，Django 本身并不支持它，但我有一个使用 Daphne(也在 Django channel 中使用)的解决方案。
创建自己的 StreamingHttpResponse能够从异步方法中检索数据流并将其提供给 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

此外，您需要使用 Daphne 运行您的 Django 网络服务器。正确支持服务器发送事件 (SSE)。它得到“Django 软件基金会”的官方支持，语法与 gunicorn 相似。 ，但使用 asgi.py而不是 wsgi.py .
要使用它 - 您可以使用以下方式安装: pip install daphne并更改命令: python manage.py runserver类似于: daphne -b 0.0.0.0 -p 8000 sse_demo.asgi:application .
不确定它是否适用于 gunicorn .
如果您还有其他问题，请告诉我。