如何在Starlette中间件中检查响应类型？两大疑问求解

问题背景 #

我正在创建一个中间件，逻辑依赖响应类型：

• 对于Response、HTMLResponse、JSONResponse这类即时响应，需要在响应发送后关闭数据库会话；
• 对于StreamingResponse则无需关闭，要等流结束后再处理连接关闭（打算在生成器中实现）。

但遇到的问题是：无论路由里设置哪种响应类型，打印type(response)得到的始终是starlette.middleware.base._StreamingResponse，导致无法通过类型判断来执行不同逻辑。

我的疑问：

1. 为什么中间件总是把响应包装成_StreamingResponse类？
2. 有没有办法检测到真实的响应类型？

目前只能通过指定无需关闭会话的端点路径来临时解决，但这种方式扩展性差，新增流端点时需要手动添加路径。

代码示例 #

_engine = create_async_engine(
    f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{database}"
)
AsyncSessionFactory = async_sessionmaker(bind=_engine, expire_on_commit=False)


class DBSessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request.state.db_session_factory = AsyncSessionFactory
        request.state.active_session = None

        response = await call_next(request)
        debug_logger.debug(f'response type: {type(response)}')  # 始终输出 `_StreamingResponse`

        if request.state.active_session and not isinstance(response, StreamingResponse):
            request.state.active_session.close()   
            
        return response


# 从工厂获取或创建会话
async def get_db_session(request: Request) -> AsyncSession:
    if hasattr(request.state, "active_session") and request.state.active_session:
        session = request.state.active_session
    elif hasattr(request.state, "db_session_factory"):
        request.state.active_session = request.state.db_session_factory()
        session = request.state.active_session
        debug_logger.debug(f'session {id(session)} opened')
    else:
        session = AsyncSessionFactory()
    return cast(AsyncSession, session)

解答 #

1. 为什么响应会被包装成_StreamingResponse？ #

这是Starlette的BaseHTTPMiddleware的设计特性：为了统一处理所有响应类型，它会把非流式响应（比如Response、JSONResponse）自动包装成_StreamingResponse（StreamingResponse的内部子类）。这样中间件可以用统一的方式处理响应发送逻辑，无需区分不同响应类型的实现细节。

2. 如何检测真实的响应类型？ #

可以通过访问响应对象的__wrapped__属性来获取原始响应——BaseHTTPMiddleware在包装响应时会保留原始对象的引用在这个属性里。

修改中间件里的判断逻辑即可：

# 获取真实的响应对象
real_response = response.__wrapped__ if hasattr(response, "__wrapped__") else response

if request.state.active_session and not isinstance(real_response, StreamingResponse):
    request.state.active_session.close()

另外，也可以通过检查响应的body_iterator属性辅助判断：流式响应的body_iterator是生成器/异步生成器，而即时响应的body_iterator通常是包含完整响应内容的列表。不过优先使用__wrapped__的方式更直接可靠。

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