FastAPI中如何为同一端点添加不同路由类的多个路由器？

问题场景 #

现有一个 /users 路由，初始化默认路由器时指定了用于翻译的 LocalizationRoute：

api_router = APIRouter(
    responses={},
    route_class=LocalizationRoute,
    dependencies=[],
)

现在需要为该路由添加日志功能，已实现基于 APIRoute 的 LoggingRoute 类：

def log_info(req_body, res_body):
    print(req_body)
    print(res_body)


class LoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            req_body = await request.body()
            response = await original_route_handler(request)

            if isinstance(response, StreamingResponse):
                res_body = b''
                async for item in response.body_iterator:
                    res_body += item

                task = BackgroundTask(log_info, req_body, res_body)
                return Response(content=res_body, status_code=response.status_code,
                                headers=dict(response.headers), media_type=response.media_type, background=task)
            else:
                res_body = response.body
                response.background = BackgroundTask(log_info, req_body, res_body)
                return response

        return custom_route_handler

尝试通过 api_router.include_router(APIRouter(router_class=LoggingRoute)) 挂载日志路由器，但未生效，端点仍使用 LocalizationRoute，因此疑问：FastAPI 是否支持为同一端点添加不同路由类的多个路由器？

解答 #

FastAPI不支持为同一端点直接叠加多个路由类。原因是：路由器的 route_class 是实例级属性，当用 include_router 挂载子路由器时，子路由器的 route_class 会被忽略，所有端点都会沿用父路由器的路由类生成处理逻辑。

可行解决方案 #

1. 组合路由类功能

将 LocalizationRoute 和 LoggingRoute 的逻辑合并到一个自定义路由类中，通过继承实现功能叠加：

def log_info(req_body, res_body):
    print(req_body)
    print(res_body)

class LoggingLocalizationRoute(LocalizationRoute):
    def get_route_handler(self) -> Callable:
        # 先获取本地化路由的原始处理器
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            req_body = await request.body()
            response = await original_route_handler(request)

            if isinstance(response, StreamingResponse):
                res_body = b''
                async for item in response.body_iterator:
                    res_body += item

                task = BackgroundTask(log_info, req_body, res_body)
                return Response(content=res_body, status_code=response.status_code,
                                headers=dict(response.headers), media_type=response.media_type, background=task)
            else:
                res_body = response.body
                response.background = BackgroundTask(log_info, req_body, res_body)
                return response

        return custom_route_handler

然后用这个组合类初始化路由器：

api_router = APIRouter(
    responses={},
    route_class=LoggingLocalizationRoute,
    dependencies=[],
)

2. 改用中间件实现日志

如果日志功能不需要与路由类强绑定，可使用 FastAPI 的中间件实现，与 LocalizationRoute 共存：

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        req_body = await request.body()
        response = await call_next(request)
        
        # 处理流式响应
        if isinstance(response, StreamingResponse):
            res_body = b''
            async for chunk in response.body_iterator:
                res_body += chunk
            # 重新构建响应
            response = Response(
                content=res_body,
                status_code=response.status_code,
                headers=dict(response.headers),
                media_type=response.media_type
            )
        else:
            res_body = response.body
        
        # 后台执行日志任务
        response.background = BackgroundTask(log_info, req_body, res_body)
        return response

# 给目标路由器添加中间件
api_router.add_middleware(LoggingMiddleware)

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