FastAPI响应迭代器阻塞异步函数/后台任务问题求助

问题原因 #

你用[section async for section in response.body_iterator]这种方式会一次性耗尽整个响应体迭代器，迫使FastAPI必须等所有响应内容生成完毕（包括触发后台任务的路由逻辑完全执行完）才会返回响应。而FastAPI的后台任务默认是在响应发送给客户端后才启动的，所以就出现了后台任务被阻塞的情况。

解决方案 #

不要直接一次性读取全部响应体，而是创建一个包装的异步生成器，在流式传输响应的同时收集响应体内容，这样既能拿到响应体，又不会阻塞后台任务的执行。

完整代码示例 #

from fastapi import FastAPI, Request, Response
import asyncio
from typing import AsyncGenerator

app = FastAPI()

async def wrap_body_iterator(iterator: AsyncGenerator[bytes, None], body_list: list):
    async for chunk in iterator:
        body_list.append(chunk)
        yield chunk

@app.middleware("http")
async def log_response_body(request: Request, call_next):
    response: Response = await call_next(request)
    
    # 用列表收集响应体片段
    response_body = []
    # 包装原有的body_iterator，流式传输的同时收集内容
    response.body_iterator = wrap_body_iterator(response.body_iterator, response_body)
    
    # 异步处理收集到的响应体，避免阻塞主流程
    async def process_collected_body():
        await asyncio.sleep(0.1)  # 等待响应流式传输完成
        full_body = b"".join(response_body).decode("utf-8")
        print(f"完整响应体: {full_body}")
    
    # 启动后台任务处理响应体，不影响主响应发送流程
    asyncio.create_task(process_collected_body())
    
    return response

# 带后台任务的测试路由
@app.get("/")
async def test_route(response: Response):
    async def demo_background_task():
        await asyncio.sleep(2)
        print("后台任务执行完成")
    
    # 给响应绑定后台任务
    response.background_tasks.add_task(demo_background_task)
    
    return {"status": "success", "message": "Hello World"}

关键说明 #

• wrap_body_iterator作为包装器，一边迭代原响应体片段、将其返回给客户端保证流式传输，一边把片段存入列表，不会等待全部响应内容生成。
• 收集到的响应体片段通过异步后台任务处理，完全不影响主响应的发送流程，FastAPI会正常返回响应后再启动路由绑定的后台任务。

内容的提问来源于stack exchange，提问作者Ammar Ahmad Khan