如何从httpx.stream获取响应头与AsyncIterable内容返回FastAPI StreamingResponse？

问题本质 #

你碰到的httpx.StreamClosed错误，是因为httpx的响应上下文管理器在FastAPI启动流式传输前就已退出，导致响应流被提前关闭。FastAPI的StreamingResponse会异步迭代返回数据，但如果httpx的响应对象已经被上下文销毁，就会触发这个错误。

解决方案 #

用异步生成器包裹httpx的请求逻辑，让httpx的响应对象在整个流式传输过程中保持打开状态，同时可以轻松获取响应头信息。

完整代码示例 #

from fastapi import FastAPI, StreamingResponse
import httpx

app = FastAPI()

async def file_stream_generator(target_url: str):
    # 将httpx请求逻辑放在生成器内部，确保上下文覆盖整个流式过程
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", target_url) as resp:
            # 提取并处理响应头，示例：获取内容类型、文件名
            content_type = resp.headers.get("Content-Type", "application/octet-stream")
            content_disposition = resp.headers.get("Content-Disposition")
            filename = content_disposition.split("filename=")[-1].strip('"') if content_disposition else "unknown-file"
            
            # 这里可以根据业务需求处理头信息，比如打印日志、存储到数据库等
            print(f"目标文件信息：类型={content_type}, 文件名={filename}")
            
            # 逐块读取并返回流数据
            async for chunk in resp.aiter_bytes(chunk_size=4096):
                yield chunk

@app.get("/proxy-download")
async def proxy_download():
    # 替换为实际的文件下载地址
    source_url = "https://example.com/large-dataset.csv"
    
    response = StreamingResponse(
        file_stream_generator(source_url),
        media_type="application/octet-stream"
    )
    # 可选：将目标服务器的文件名头传递给客户端
    response.headers["Content-Disposition"] = f'attachment; filename="downloaded-file.csv"'
    return response

核心要点 #

• 把httpx的client.stream上下文放在异步生成器内部，确保响应流在整个数据传输周期内都处于打开状态，直到所有数据块发送完毕。
• 在生成器中可以直接访问httpx响应的headers属性，满足你同时获取头信息和文件数据的需求。
• 使用aiter_bytes()按块读取数据，避免内存过载，符合流式传输的设计目的。

备选方案：自定义异步上下文管理器 #

如果需要更精细控制资源生命周期，可以用自定义上下文管理器确保httpx资源被正确回收：

from fastapi import FastAPI, StreamingResponse
import httpx
from contextlib import asynccontextmanager

app = FastAPI()

@asynccontextmanager
async def open_httpx_stream(url: str):
    client = httpx.AsyncClient()
    resp = None
    try:
        resp = await client.stream("GET", url)
        yield resp
    finally:
        if resp:
            await resp.aclose()
        await client.aclose()

async def stream_data(resp):
    async for chunk in resp.aiter_bytes():
        yield chunk

@app.get("/direct-download")
async def direct_download():
    source_url = "https://example.com/large-file.zip"
    async with open_httpx_stream(source_url) as resp:
        # 处理响应头
        media_type = resp.headers.get("Content-Type", "application/octet-stream")
        return StreamingResponse(
            stream_data(resp),
            media_type=media_type,
            headers={"Content-Disposition": resp.headers.get("Content-Disposition", "")}
        )

这种方式通过上下文管理器明确控制httpx客户端和响应的生命周期，避免资源泄漏，同时保证流在传输期间有效。

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