FastAPI中如何流式返回JSON格式的分片数据？

核心问题 #

1. StreamingResponse不自动序列化Python对象：直接yield字典data时，FastAPI的StreamingResponse无法将Python字典自动序列化为JSON字节流——它仅接受字符串、字节或异步生成器返回的这类可直接传输的内容，不会帮你做对象序列化。
2. 纯JSON对象拼接不合法：即使把每个字典单独序列化为JSON字符串后yield，多个独立的{"iteration":x}直接拼接在一起，并不是合法的JSON结构（JSON不允许多个顶级对象），前端解析时会报错。

两种可行解决方案 #

方案1：使用JSON Lines格式（推荐） #

JSON Lines（每行一个JSON对象）是流式JSON场景的标准格式，media_type设为application/jsonlines，每个分片是独立的JSON字符串加换行符，前端可以逐行解析：

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

app = FastAPI()

@app.get("/experimental/stream", summary="Streaming JSON Lines")
async def stream_response():
    async def stream_generator():
        for i in range(10):
            data = {"iteration": i}
            # 序列化后加换行符，符合JSON Lines规范
            yield json.dumps(data) + "\n"
            await asyncio.sleep(1)
    return StreamingResponse(stream_generator(), media_type="application/jsonlines")

方案2：流式输出合法JSON数组 #

如果需要返回标准的JSON数组（前端可直接解析为数组），可以构建流式数组：先输出[，然后逐个输出带逗号的JSON元素，最后输出]：

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

app = FastAPI()

@app.get("/experimental/stream-array", summary="Streaming JSON Array")
async def stream_response():
    async def stream_generator():
        total = 10
        # 先输出数组开头
        yield "["
        for i in range(total):
            data = {"iteration": i}
            json_str = json.dumps(data)
            # 最后一个元素不加逗号
            if i < total - 1:
                yield json_str + ", "
            else:
                yield json_str
            await asyncio.sleep(1)
        # 输出数组结尾
        yield "]"
    return StreamingResponse(stream_generator(), media_type="application/json")

补充说明 #

• 方案1的JSON Lines更适合纯流式场景，前端可以实时处理每一行数据；
• 方案2的数组格式更符合传统JSON解析习惯，但需要处理逗号和首尾括号的拼接逻辑；
• 异步函数里避免用time.sleep()，它会阻塞事件循环，推荐用asyncio.sleep()替代。

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