StreamingResponse is returning all content at once

[LeonardoSanBenitez]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
import time

app = FastAPI()

async def fake_json_streamer():
    t0 = time.time()
    for i in range(10):
        print(f"Chunk being yielded (time {int((time.time()-t0)*1000)}ms)", flush=True)
        yield json.dumps( {"message": "Hello World"}) + '\n'
        time.sleep(0.5)
    print(f"Over (time {int((time.time()-t0)*1000)}ms)", flush=True)

@app.get("/test")
async def test():
    return StreamingResponse(fake_json_streamer(), media_type='text/event-stream')

Description

I'm using StreamingResponse to return json content, and I would like to see each new "chunck" (complete json) being returned every 0.5s.
However, when I access the FastAPI endpoint, I'm receiving all the content at once (after the complete 5s needed to stream all content).

That is, when I access the endpoint with the following client code:

import requests
import time
t0 = time.time()
response = requests.get('http://localhost:8080/test')
for chunk in response.iter_lines():
    print(f"Chunk (time {int((time.time()-t0)*1000)}ms): {chunk}")

I see the output (in the client):

Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'
Chunk (time 5026ms): b'{"message": "Hello World"}'

And on the server logs I see:

INFO:     172.21.0.1:55586 - "GET /test HTTP/1.1" 200 OK
Chunk being yielded (time 0ms)
Chunk being yielded (time 500ms)
Chunk being yielded (time 1001ms)
Chunk being yielded (time 1502ms)
Chunk being yielded (time 2002ms)
Chunk being yielded (time 2503ms)
Chunk being yielded (time 3004ms)
Chunk being yielded (time 3505ms)
Chunk being yielded (time 4005ms)
Chunk being yielded (time 4506ms)
Over (time 5006ms)

That is, the chuncks of json content are being generated and yielded roughly every 500ms, but the client only see the the result at 5026ms.

If this is a problem with the client code (requests library), I apology for the misunderstanding.

Operating System

Linux

Operating System Details

Ubuntu 20.04.6 LTS

FastAPI Version

0.99.1

Pydantic Version

1.10.13

Python Version

3.8.10

Additional Context

This is a simplified example that I hope it's easy to reproduce.
My ultimate goal is to return the response of openai.Completion.create (from the openai library) with stream=True, so any example/tutorial/code in that direction would also be helpful.

[nymous]

Hello,
There is indeed a mistake in your client code. The requests documentation for streaming requests indicates that you must pass the stream=True argument to the get() method.
If you do this you will see that the client prints every chunk as it receives it.

You can also check it with curl, for example curl http://localhost:8000/test would have shown you that the response was streamed correctly.

Do note that your code is not using async functions properly: in your async def fake_json_streamer() function you use the sync time.sleep() function, which will block the event loop and the entire application because it is executed in an async context. This will prevent requests to be served concurrently. You should not mark this function as async and keep it a normal def. This will also mean that the endpoint must not be marked async as it is calling a blocking function, so you should also keep it a normal def test().
Another option would be to use the async equivalent of the blocking functions you need, such as asyncio.sleep() in your case.

[LeonardoSanBenitez]

Unfortunately, both curl http://localhost:8080/test and requests.get with stream=True show the same wrong behavior: everything is returned just after 5s.
Screen recording of the curl option:

For requests, the output is the same as in the message above.

About the usage of async: thank you for noting, I've updated the code to use asyncio.sleep().

[nymous]

That's weird, I don't have any issue with this code

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

import json
import time
import asyncio

app = FastAPI()

async def fake_json_streamer():
    t0 = time.time()
    for i in range(10):
        print(f"Chunk being yielded (time {int((time.time()-t0)*1000)}ms)", flush=True)
        yield json.dumps( {"message": "Hello World"}) + '\n'
        await asyncio.sleep(0.5)
    print(f"Over (time {int((time.time()-t0)*1000)}ms)", flush=True)

@app.get("/test")
async def test():
    return StreamingResponse(fake_json_streamer(), media_type='text/event-stream')

fastapi-streaming-response.mp4

[LeonardoSanBenitez]

Weird... Are you using the same version as me?
And thank you for the support :)

[Kludex]

Missing stream=True on requests.get(..., stream=True).

https://requests.readthedocs.io/en/latest/api/

[VincentRoma]

This is not the correct answer and shouldn't be marked as it.

[LeonardoSanBenitez]

I'm sorry @Kludex, but I still see the same problem:

Is it possible that the problem is not related to FastAPI?
If so, I'll close the issue, but do you have any suggestion about how to debug the actual problem?

[Kludex]

The problem is not with FastAPI. You can try other HTTP clients.

[LeonardoSanBenitez]

Thanks, I'm closing the question

[anantgupta129]

hi @LeonardoSanBenitez hope you have resolved the issue, i faced same issue currently and was able to solve it by using non asynchronous generator function, so in your case

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

import json
import time
import asyncio

app = FastAPI()

def fake_json_streamer():
    t0 = time.time()
    for i in range(10):
        print(f"Chunk being yielded (time {int((time.time()-t0)*1000)}ms)", flush=True)
        yield json.dumps( {"message": "Hello World"}) + '\n'
        time.sleep(0.5)
    print(f"Over (time {int((time.time()-t0)*1000)}ms)", flush=True)

@app.get("/test")
async def test():
    return StreamingResponse(fake_json_streamer(), media_type='text/event-stream')

which worker for me, but don't have the clear explanation
this might be useful

Second, if you are executing a blocking operation (i.e., blocking I/O-bound or CPU-bound tasks) inside the StreamingResponse's generator function, you should define the generator function with def instead of async def, as, otherwise, the blocking operation, as well as the time.sleep() function that is used inside your generator, would block the event loop.

[anantgupta129]

secondly you should be suing await asyncio.sleep(0.5) instead of time.sleep() with a async generator

[vskritsk]

Compression middleware leads to such a pattern. If that's the case, try removing it from the streaming endpoint and setting it per each router.

[LeonardoSanBenitez]

Thank you, but that was not the case, no middleware was used

[asaf]

@LeonardoSanBenitez are you running through some proxy? if yes that's probably the reason, some proxies don't proxies SSE properly!

[LeonardoSanBenitez]

No, I was running FastAPI locally and without anything in the middle.
I tested the code in a different computer and it worked as expected, so clearly it's not a problem with FastAPI itself.
I could not pinpoint what difference in the two computers configurations lead to this behavior, and as it worked well in production I didn't investigated the subject further :)

[MiesnerJacob]

Ever find a solution here? I am running into the same problem. I think it may be related to the use of an async generator as the function the route is streaming. Although I am unable to find a solution.

[lyris]

I have also observed a similar issue, whereby even when setting stream=True, the HTTP client only receives a batch of data when the sum of chunk reaches 64 KB.

[LeonardoSanBenitez]

No, I have never clarified what was/is going wrong.
I have the same code with the same versions working on the computer of a colleague, so I suspect I can be something dependent on the operating system (he was using windows, I was using ubuntu). It working as well in the production server (alpine).
But I have never investigated further

[qkxie]

I have met the same problem and find the real reason of my case. Because I have a Gzip Middleware, this middleware cause the stream response not work. When I disable Gzip middleware, everything works well.

[La-fe]

use iter_content

[anantgupta129]

#10701 (reply in thread)

[idailylife]

Thanks! I finally found out that my http response logging middleware is the culprit...