StreamingResponse doesn't support PEP-492 asynchronous iterators

[Korijn]

PEP-492 asynchronous iterators have the following signature:

class AsyncIterable:
    def __aiter__(self):
        return self

    async def __anext__(self):
        data = await self.fetch_data()
        if data:
            return data
        else:
            raise StopAsyncIteration

For example, the Azure storage python library returns such an asynchronous iterator for blob downloading. I can't pass it to Starlette though; it will execute in the threadpool and crash there:

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/uvicorn/protocols/http/httptools_impl.py", line 385, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
    return await self.app(scope, receive, send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/fastapi/applications.py", line 142, in __call__
    await super().__call__(scope, receive, send)  # pragma: no cover
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/applications.py", line 134, in __call__
    await self.error_middleware(scope, receive, send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/middleware/errors.py", line 178, in __call__
    raise exc from None
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/middleware/errors.py", line 156, in __call__
    await self.app(scope, receive, _send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/exceptions.py", line 73, in __call__
    raise exc from None
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/exceptions.py", line 62, in __call__
    await self.app(scope, receive, sender)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/routing.py", line 590, in __call__
    await route(scope, receive, send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/routing.py", line 208, in __call__
    await self.app(scope, receive, send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/routing.py", line 44, in app
    await response(scope, receive, send)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/responses.py", line 196, in __call__
    async for chunk in self.body_iterator:
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/concurrency.py", line 45, in iterate_in_threadpool
    yield await run_in_threadpool(_next, iterator)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/concurrency.py", line 25, in run_in_threadpool
    return await loop.run_in_executor(None, func, *args)
  File "/home/korijn/.pyenv/versions/3.6.8/lib/python3.6/concurrent/futures/thread.py", line 56, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/contextvars/__init__.py", line 38, in run
    return callable(*args, **kwargs)
  File "/home/korijn/.local/share/virtualenvs/aiotest--vQHOsY--/home/korijn/.pyenv/shims/python/lib/python3.6/site-packages/starlette/concurrency.py", line 37, in _next
    return next(iterator)
TypeError: '_AsyncChunkIterator' object is not an iterator

I have to wrap the iterator in an asynchronous generator function for it to work:

    async def chunks_generator():
        async for i in blob.chunks():
            yield i

    return StreamingResponse(chunks_generator())

[tomchristie]

Okay, I guess we'd need to change this test...

starlette/starlette/responses.py

Line 180 in 5c43dde

if inspect.isasyncgen(content):

Any idea what the check ought to be instead?

[Korijn]

I don't think there is builtin function we can call. The glossary says:

asynchronous iterator
An object that implements the __aiter__() and __anext__() methods. __anext__ must return an awaitable object. async for resolves the awaitables returned by an asynchronous iterator’s __anext__() method until it raises a StopAsyncIteration exception. Introduced by PEP 492.

So we could check if the object defines those methods?

[witling]

I need this issue to be fixed as well. Why not just go with something like:

if hasattr(content, '__aiter__') and hasattr(content, '__anext__'):
    # content is an async iterator
    pass