FASTAPI中间件提取响应流内容时不影响原响应的方案咨询

问题原因 #

response.body_iterator是一次性异步迭代器，当你用async for遍历完后，迭代器内的内容就被耗尽了，但原响应的Content-Length头部依然存在，导致返回时出现RuntimeError("Response content shorter than Content-Length")错误。

优化方案 #

方案1：包装响应体迭代器，保留原流 #

这种方法无需重新创建Response，读取内容后把读取到的片段重新包装成新的异步迭代器，替换原响应的body_iterator，既拿到响应内容，又不影响原响应正常返回：

from fastapi import Request
import json
import asyncio

@app.middleware("http")
async def publish_to_event_stream(request: Request, call_next):
    response = await call_next(request)
    resp_content = None
    content_type = response.headers.get("content-type", "")
    
    try:
        # 兼容带charset的JSON类型，比如application/json; charset=utf-8
        if "application/json" in content_type:
            response_body = b""
            chunks = []
            # 读取所有响应片段
            async for chunk in response.body_iterator:
                chunks.append(chunk)
                response_body += chunk
            
            resp_content = json.loads(response_body.decode())
            
            # 重新构建异步迭代器，把片段放回
            async def new_body_iterator():
                for chunk in chunks:
                    yield chunk
                    await asyncio.sleep(0)  # 符合异步迭代器的控制权让出规范
            
            # 替换原响应的迭代器
            response.body_iterator = new_body_iterator()
    except Exception as e:
        logger.error('response_parsing_failure', exc_info=e, response=response)
    
    return response

方案2：完善的重新创建Response方法 #

如果选择重新构建Response，要保留原响应的状态码、头部等元数据，避免丢失重要信息：

from fastapi import Request, Response
import json

@app.middleware("http")
async def publish_to_event_stream(request: Request, call_next):
    response = await call_next(request)
    resp_content = None
    content_type = response.headers.get("content-type", "")
    
    try:
        if "application/json" in content_type:
            response_body = b""
            async for chunk in response.body_iterator:
                response_body += chunk
            
            resp_content = json.loads(response_body.decode())
            
            # 复制原响应头部，移除旧的Content-Length（新Response会自动计算）
            headers = dict(response.headers)
            headers.pop("Content-Length", None)
            
            # 重新创建响应，保留原状态码和头部
            response = Response(
                content=response_body,
                status_code=response.status_code,
                headers=headers,
                media_type=content_type
            )
    except Exception as e:
        logger.error('response_parsing_failure', exc_info=e, response=response)
    
    return response

方案对比 #

• 方案1更优雅，完整保留原响应的所有属性，适合无需修改响应内容的场景。
• 方案2适合需要调整响应内容的场景，但需注意处理头部信息，避免遗漏。

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