FastAPI流式响应中aiohttp连接提前关闭问题求助

你的错误根源在于上下文管理器的生命周期和流式响应的迭代周期不匹配：当proxy_stream函数返回StreamingResponse后，外层的async with ClientSession()和async with http_session.post()会立即执行退出逻辑，关闭aiohttp的连接和响应对象。但此时process_response生成器还在尝试读取响应内容，自然会抛出Connection closed错误。

将HTTP请求的上下文管理逻辑直接嵌入异步生成器中，让连接和响应对象在整个流式传输过程中保持打开状态，直到所有数据块发送完毕才关闭。

修改后的代码示例：

from fastapi import APIRouter
from fastapi.responses import StreamingResponse
import aiohttp

router = APIRouter()

# 封装外部API的流式请求与处理逻辑
async def stream_external_data():
    async with aiohttp.ClientSession() as http_session:
        async with http_session.post(
            "https://streaming-api",
            json={"json": "body"}
        ) as response:
            # 确保外部API返回成功响应
            response.raise_for_status()
            async for chunk in response.content.iter_chunked(128):
                # 处理数据块后返回
                yield process_chunk(chunk)

@router.post("/")
async def proxy_stream():
    # 返回流式响应，传入封装好的异步生成器
    return StreamingResponse(
        stream_external_data(),
        media_type="application/octet-stream"  # 根据实际数据类型调整
    )

• 如果process_chunk是耗时的同步操作，建议改为异步函数（async def process_chunk(chunk)），避免阻塞FastAPI的事件循环。
• 添加响应状态码校验（response.raise_for_status()），可以提前捕获外部API的错误响应，避免无效的流式传输。
• 根据外部API返回的数据格式，设置正确的media_type参数（比如JSON流设为application/json，文本流设为text/plain）。

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