FastAPI 自定义 Request 与 APIRoute 类：深度接管请求处理链路与响应修饰

2026-09-06 16:29:40作者：管翌锬

本文围绕 FastAPI 官方文档 Custom Request and APIRoute class 展开，讲解如何通过重写 Request 子类与 APIRoute 子类来拦截、转换请求体（如解压 gzip 请求、在非 JSON 编码下工作），并在响应中注入自定义行为（如响应耗时头、异常时读取原始请求体）。读完本文，你将掌握：Request/APIRoute 两个扩展点的正确打开方式、ASGI 中 scope 与 receive 的角色，以及 APIRouter 的 route_class 参数如何按路由粒度生效。文末结合 fastapi/routing.py 源码说明这些扩展点在框架内部的真实调用链，帮助你在生产环境中安全使用这些"进阶"能力。

需要特别说明：文档将其标记为 "advanced"（进阶）特性，如果你刚开始接触 FastAPI，可以先了解概念，不必立即上手。

适用场景：为什么需要自定义 Request 和 APIRoute

有些需求无法用普通中间件优雅地解决，此时重写 Request 和 APIRoute 的逻辑是一个好替代方案。文档列举的典型用例包括：

将非 JSON 的请求体转换为 JSON（例如 msgpack）；
解压 gzip 压缩的请求体；
自动记录（日志化）所有请求体。

这类需求的共同点是：需要在请求体被 FastAPI 框架本身解析之前，就读取或改写请求体。中间件方案同样可以做到，而自定义 Request/APIRoute 则把转换逻辑封装在路由层，粒度更精确，且与 OpenAPI、依赖注入等机制保持完全一致。

实战一：用自定义 `GzipRequest` 支持 gzip 压缩请求体

完整可运行示例见 docs_src/custom_request_and_route/tutorial001_an_py310.py。

提示：文档说明这是一个"玩具级"演示。如果你在生产环境需要 gzip 支持，可以直接使用框架内置的 GzipMiddleware，无需手写这一套。

第一步：重写 `Request.body()` 的 `GzipRequest` 类

核心思路是覆写 Request.body() 方法：先取出原始字节，若请求头 Content-Encoding 中包含 gzip，就解压后再返回。这样同一个路由既接受压缩请求也接受普通请求：

class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body

实现要点：

super().body() 调用父类逻辑读取原始请求体字节；
hasattr(self, "_body") 做缓存，保证请求体只读取/解压一次（Starlette 的 Request.body() 本身也有 _body 缓存约定，这里延续了同样的模式）；
用 headers.getlist("Content-Encoding") 判断编码，没有 gzip 时原样返回，实现"透明降级"。

第二步：重写 `get_route_handler()` 的 `GzipRoute` 类

接下来创建一个 fastapi.routing.APIRoute 的子类，让它使用上面定义的 GzipRequest。这次覆写的是 APIRoute.get_route_handler() 方法——该方法返回一个函数，这个函数才是真正"接收请求、返回响应"的路由处理器：

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

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler

然后在应用级别启用它（app.router 是 APIRouter，修改其 route_class 属性后新注册的路由都会使用该类）：

app = FastAPI()
app.router.route_class = GzipRoute

@app.post("/sum")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

技术细节：`scope` 与 `receive` 从何而来

文档特别强调了一段 ASGI 层面的原理：

每个 Request 都有 request.scope 属性，它只是一个包含请求元数据的 Python dict；
Request 还有 request.receive，这是一个用于"接收"请求体的函数；
scope 字典与 receive 函数都是 ASGI 规范的一部分，两者正是构造一个新的 Request 实例所需的全部输入。

所以 GzipRoute 的处理器中唯一"不同"的动作，就是把普通 Request 用 (request.scope, request.receive) 重新包装成 GzipRequest，再交还给原始处理器。后续的依赖解析、参数校验、响应序列化等处理逻辑完全不变；只是由于 GzipRequest.body() 被重写，当 FastAPI 需要加载请求体（解析 body 参数）时，数据会自动先经过 gzip 解压。

这一点从源码可以直接印证：路由处理器最终由 get_request_handler 生成，其中读取请求体走的正是 body_bytes = await request.body()（见 fastapi/routing.py）。因此只要"喂"进路由处理器的是 GzipRequest 实例，body 解析链路就会自动复用你覆写后的 body() 方法——这正是整个方案能生效的关键。

实战二：在异常处理器中访问请求体

文档给出的第二个例子，展示如何借助自定义 APIRoute，在捕获校验异常时读取原始请求体并放入错误响应。完整示例见 docs_src/custom_request_and_route/tutorial002_an_py310.py：

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

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler

app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute

@app.post("/")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return sum(numbers)

要点：

只需把对原始路由处理器的调用包进 try/except 块；
发生异常时，Request 实例仍然在作用域内，因此可以 await request.body() 读取（并复用已缓存的）请求体，将其与校验错误一起放进 422 响应的 detail 中。

文档同时提醒：如果只是想在校验失败时拿到请求体，更简单的方式是在 RequestValidationError 的自定义异常处理器里直接使用异常对象自带的 body 属性（参见 Handling Errors 文档）。本例的价值在于演示"如何与框架内部组件交互"这一通用模式。

实战三：为某个 `APIRouter` 单独指定 `route_class`

自定义路由类不必全局生效，APIRouter 提供了 route_class 参数（源码中默认值就是 APIRoute，见 fastapi/routing.py）：

import time
from collections.abc import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


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

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=TimedRoute)

@app.get("/")
async def not_timed():
    return {"message": "Not timed"}

@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}

app.include_router(router)

完整示例见 docs_src/custom_request_and_route/tutorial003_py310.py。

在这个示例中：

挂在 router 下的 path operation 会使用自定义的 TimedRoute 类，响应里会多出一个 X-Response-Time 头，记录生成响应所花费的时间；
直接注册在 app 上的 @app.get("/") 则不受影响（返回 "Not timed"）。

这说明 route_class 的生效粒度是每个 router 独立的：APIRouter 在注册路由时会用 route_class 实例化路由对象（源码中为 route_class = route_class_override or self.route_class，见 fastapi/routing.py），因此不同 router 可以携带不同的路由行为，便于按模块做性能埋点、审计、限流等差异化处理。

源码纵览：扩展点在 FastAPI 内部如何被调用

结合 fastapi/routing.py 的实现，可以把这套机制串成一条清晰的调用链：

路由初始化时绑定处理器：APIRoute.__init__ 结束时执行 self.app = request_response(self.get_route_handler())（第 1223 行）。因为 Python 的多态机制，如果你的子类重写了 get_route_handler()，这里拿到的就是子类版本——这就是为什么"只覆写一个方法"就能改变整个路由行为。
get_route_handler() 返回真正的请求处理器：默认实现（第 1225-1249 行）调用 get_request_handler(...)，把路由的依赖 dependant、body 字段、响应模型等配置打包成一个 async def app(request) -> Response 协程函数，负责读取 body、解析依赖、执行端点函数、序列化响应。
子类通过"装饰"原始处理器介入：文档中所有例子都遵循同一模式——先 original_route_handler = super().get_route_handler() 拿到原始处理器，再返回一个闭包 custom_route_handler，在调用前后插入自定义逻辑（换请求对象、包 try/except、加响应头）。这种"包裹"模式保证了你无需理解框架内部细节，只需围绕原始处理器加一层。
请求体读取是唯一的耦合点：get_request_handler 生成的处理器中，body 的读取是 await request.body()（第 433 行）和 await request.form()（第 430 行）。因此 GzipRequest 只需覆写 body()，就能让后续所有 JSON 解析、依赖注入透明地拿到解压后的数据。
应用级启用方式：app.router.route_class = GzipRoute 之所以有效，是因为 FastAPI 内部的默认 router 就是 APIRouter，其 route_class 默认为 APIRoute（第 2414 行），修改属性后，之后 add_api_route 时实例化的便是 GzipRoute。

从源码结构看，APIRoute.handle() 中还处理了路由方法不匹配（405）、root_path 前缀等细节，但这些都发生在"你的处理器"外层，不影响上述扩展模式。

验证：配套测试用例

仓库为这些示例提供了自动化测试，例如 tests/test_tutorial/test_custom_request_and_route/test_tutorial001.py 覆盖了 gzip 请求类的教程示例，可用于验证"发送 gzip 压缩的 body 后，/sum 路由能正确返回求和结果"这类端到端行为。运行测试前需按 pyproject.toml 安装依赖，仓库脚本 scripts/test.sh 展示了项目使用的测试入口。

小结

两个扩展点：重写 Request.body() 处理"请求体进什么"，重写 APIRoute.get_route_handler() 处理"谁拿到请求、响应如何修饰"，二者组合覆盖文档列举的 gzip 解压、body 日志、非 JSON 编码转换等场景。
ASGI 基础：(scope, receive) 是构造 Request 的全部原料，理解这一点后，自定义请求类的原理一目了然。
生效粒度灵活：app.router.route_class = ... 全局生效，APIRouter(route_class=...) 按路由组生效，二者可混用。
更简单的替代方案优先：gzip 解压可直接用内置 GzipMiddleware；在 422 响应中携带请求体可直接用 RequestValidationError 的 body 属性。只有当需求必须深入路由处理器内部时，本文的 APIRoute 重写模式才值得采用。