How can I chain request.stream() to StreamingResponse ?

[nu774]

async def echo(request):
    return StreamingResponse(request.stream())

I guessed this would iteratively echo back chunked request to the client, but it doesn't.

[nu774]

It seems that symptoms varies with ASGI server implementations.
On uvicorn, the client don't get any echo backs from the server side.
On hypercorn, the client get 1/2 response from the server. I mean , if client send A, B, C, D, then the server echo backs only B and D.
Anyways, both of them are broken.

The following implementation that directly talks ASGI works as intended on both server (but not on daphne):
https://gist.github.com/nu774/559ad1032de8fe6081620cf547972cdf

[DavidArchibald]

Did you have any luck figuring out a solution? I ran into the same problem where I wanted to process request.stream() and respond with StreamingResponse.

[nu774]

It seems that you cannot, with StreamingResponse.

1. StreamingResponse.listen_for_disconnect() swallows entire request body.
2. StreamingResponse.stream_response() starts response too early (before reading from body_iterator).

In other words, StreamingResponse doesn't seem to support this kind of use case. So you need to create your own Reponse class.
The following implementation worked, but I'm honestly disappointed and lost interest in this project for not supporting this kind of very trivial use case (and no one answered my simple question).

class MyStreamingResponse(Response):
    def __init__(self, content, status_code=200, headers=None, media_type=None):
        self.content = content
        self.status_code = status_code
        self.media_type = self.media_type if media_type is None else media_type
        self.init_headers(headers)

    async def __call__(self, scope, receive, send):
        initial = True
        async for chunk in self.content:
            if initial:
                await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
                initial = False
            await send({"type": "http.response.body", "body": chunk, "more_body": True})
        await send({"type": "http.response.body", "body": b"", "more_body": False})

async def echo(request):
    return MyStreamingResponse(request.stream())

[adriangb]

This is something that should be supported, we need to fix this cc @Kludex

[adriangb]

@Kludex I think the work to detect disconnects on send() solves this, but we have to bring it into Starlette

[hughsw]

[ Whoops, not sure if this is a FastAPI issue or a Starlette issue ]

First time on Starlette forum.
I too was surprised to find that code for such a basic behavior doesn't work.

I can post a detailed issue and a simple PR for a backwards-compatible fix.

[DavidArchibald]

If you already have the diff I'd be quite interested in seeing it. Maybe a PR would help out the maintainers!

[fulviodan]

any updates on this? Thanks!

[TheBigRoomXXL]

I have the same issue.

I have an endpoint where user can upload large file (up to 20Go). I want to use response.stream to process the file progressively but I also want to use a StreamingResponse to send status update back to the client in real time (to be precise i want to use see-starlette which use StreamingResponse). But when i call resquest.stream() inside the StreamingReponse's generator, it just hangs indefenitely until the client timeout, which raise ClientDisconnect. Because of that, I must first process the file entirely before sending the response, so no real-time update and most probably the client timeout before I have finished processing its request.

I going to try the custom Response solution but integrating it with see-starlette is going to be complicated. A real fix would be nice.