如何在FastAPI中使用StreamingResponse处理bytes类型变量？

完全可以用StreamingResponse直接处理bytes类型的变量，不需要先将数据保存到本地文件。

StreamingResponse的核心要求是接收一个可迭代的字节流，我们只需要把bytes变量包装成可迭代对象即可，针对不同大小的bytes数据，有两种常见实现方式：

1. 直接返回完整bytes数据（适合小体积数据） #

如果你的bytes变量体积不大，可以直接生成包含该变量的迭代器：

from fastapi import FastAPI, StreamingResponse

app = FastAPI()

@app.get("/stream-small-bytes")
def stream_small_bytes():
    # 替换为你的bytes类型变量
    file_bytes = b"模拟的二进制文件数据..." * 1000
    
    def iter_bytes():
        yield file_bytes
    
    return StreamingResponse(iter_bytes(), media_type="application/pdf")

2. 分块返回数据（适合大体积数据） #

如果bytes变量是大文件数据，建议分块返回，避免一次性占用过多内存，逻辑和流式本地文件一致：

from fastapi import FastAPI, StreamingResponse

app = FastAPI()
# 自定义分块大小，例如10KB
CHUNK_SIZE = 1024 * 10

@app.get("/stream-large-bytes")
def stream_large_bytes():
    # 替换为你的大体积bytes变量
    large_file_bytes = b"大文件二进制内容..." * 100000
    
    def iter_large_bytes():
        # 按分块大小切割bytes并逐块返回
        for i in range(0, len(large_file_bytes), CHUNK_SIZE):
            yield large_file_bytes[i:i+CHUNK_SIZE]
    
    return StreamingResponse(iter_large_bytes(), media_type="application/pdf")

两种方式都能让客户端以流式方式接收数据，和传输本地文件的效果完全一致。

内容的提问来源于stack exchange，提问作者Sanket Wagh