Repository: encode/starlette

Issue:
StreamingResponse doesn't support PEP-492 asynchronous iterators
[PEP-492 asynchronous iterators](https://www.python.org/dev/peps/pep-0492/#asynchronous-iterators-and-async-for) have the following signature:

```python
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())
```

Discussion:
uriyyo: @witling Also, it will be great to add tests to cover this feature.
witling: @uriyyo Thanks for the notes! I've updated my pull request.
witling: blocked by https://github.com/encode/starlette/pull/1040
witling: @JayH5 Thanks for your effort! I've updated my pr.
witling: annotations were implemented
witling: blocked by https://github.com/encode/starlette/pull/1072

Patch:
diff --git a/starlette/responses.py b/starlette/responses.py
index 24e3b54..d5f2734 100644
--- a/starlette/responses.py
+++ b/starlette/responses.py
@@ -1,6 +1,5 @@
 import hashlib
 import http.cookies
-import inspect
 import json
 import os
 import stat
@@ -204,7 +203,7 @@ class StreamingResponse(Response):
         media_type: str = None,
         background: BackgroundTask = None,
     ) -> None:
-        if inspect.isasyncgen(content):
+        if isinstance(content, typing.AsyncIterable):
             self.body_iterator = content
         else:
             self.body_iterator = iterate_in_threadpool(content)


Test patch:
diff --git a/tests/test_responses.py b/tests/test_responses.py
index a28bbbf..d1d26fc 100644
--- a/tests/test_responses.py
+++ b/tests/test_responses.py
@@ -101,6 +101,44 @@ def test_streaming_response():
     assert filled_by_bg_task == "6, 7, 8, 9"
 
 
+def test_streaming_response_custom_iterator():
+    async def app(scope, receive, send):
+        class CustomAsyncIterator:
+            def __init__(self):
+                self._called = 0
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                if self._called == 5:
+                    raise StopAsyncIteration()
+                self._called += 1
+                return str(self._called)
+
+        response = StreamingResponse(CustomAsyncIterator(), media_type="text/plain")
+        await response(scope, receive, send)
+
+    client = TestClient(app)
+    response = client.get("/")
+    assert response.text == "12345"
+
+
+def test_streaming_response_custom_iterable():
+    async def app(scope, receive, send):
+        class CustomAsyncIterable:
+            async def __aiter__(self):
+                for i in range(5):
+                    yield str(i + 1)
+
+        response = StreamingResponse(CustomAsyncIterable(), media_type="text/plain")
+        await response(scope, receive, send)
+
+    client = TestClient(app)
+    response = client.get("/")
+    assert response.text == "12345"
+
+
 def test_sync_streaming_response():
     async def app(scope, receive, send):
         def numbers(minimum, maximum):