Solution to set stream response with aiohttp

[icebearrrr]

Hi everyone, i got some problems when i want to download a file from some api and save it to resp.stream.
My code is like:

import aiohttp

class StreamDownload:
    async def on_get(self, req, resp)
        async with aiohttp.ClientSession() as session:
            async with session.get('api') as response:
                resp.stream = response.content.iter_chunked(1024)

response.content.iter_chunked is some async_generator like:

async for data in response.content.iter_chunked(1024):
    yield data

My problem is when the framework read from resp.stream , the connection which created by aiohttp has been closed.

In falcon.app:

...

stream = resp.stream

...

try:
    async for data in stream:
       # The connection has been closed, so it will raise some aiohttp.Connection error
        ...

Is there any solution to deal with this situation?

Thanks a lot!

[vytas7]

Hi @icebearrrr !
One way is to use an HTTP client library's functionality that is not based on context managers, if the library in question supports such (I'm not sure if this is possible with aiohttp though). The following Gist demonstrates how to manually close a connection using another popular client, httpx: Proxying a website with Falcon ASGI + httpx.

As to your aiohttp snippet, maybe you could simply wrap it in another async generator, and assign that to resp.stream?
Something along the lines of (not tested in depth!)

import aiohttp
import falcon.asgi


class StreamDownload:
    LOGO = 'https://falcon.readthedocs.io/en/stable/_static/img/logo.svg'

    async def aiohttp_stream(self, uri):
        async with aiohttp.ClientSession() as session:
            async with session.get(uri) as response:
                # NOTE(vytas): Cannot use `yield from` here in async code.
                async for chunk in response.content.iter_chunked(1024):
                    yield chunk

    async def on_get(self, req, resp):
        resp.content_type = 'image/svg+xml'
        resp.stream = self.aiohttp_stream(self.LOGO)


app = falcon.asgi.App()
app.add_route('/logo.svg', StreamDownload())

My httpx snippet demonstrates use of a pool, which you should probably consider going to production unless there's only a very limited number of requests going on (albeit closing connections manually that way is slightly less elegant than having context blocks).
Or maybe it is possible to wrap the whole app inside an aiohttp/httpx context instead 🤔

[icebearrrr]

Thanks a lot. Finally i use the falcon.asgi.Response.schedule_sync method to solve it.