FastAPI 自定义 Request 与 APIRoute：请求体改写、路由处理器覆盖与异常场景实战-迪斯科星球

FastAPI 自定义 Request 与 APIRoute：请求体改写、路由处理器覆盖与异常场景实战

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

FastAPI 在底层把每一个路径操作（path operation）都表示为一个APIRoute实例，而APIRoute通过get_route_handler()生成真正处理请求的函数。本指南以 FastAPI 官方文档自定义 Request 与 APIRoute 为主线，讲解如何通过继承Request与fastapi.routing.APIRoute来拦截、改写请求并在异常处理器中访问请求体。读完本文，你将掌握 gzip 请求体自动解压、请求/响应的环绕处理、按路由定制行为等一整套“请求级钩子”实现方案，并理解其在中间件与异常处理器之间的定位。

⚠️ 本文属于“进阶（advanced）”话题。如果你刚开始学习 FastAPI，建议先跳过本节，待熟悉路由、依赖与中间件机制后再回来阅读。

为什么要覆盖 Request 与 APIRoute

某些场景下，你希望在业务代码执行前统一读取或操纵请求体，此时覆盖Request与APIRoute类的既有逻辑，往往是比写在中间件（middleware）里更优雅、更聚焦的替代方案。原文档给出的典型使用场景包括：

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

这类需求的共同特征是：请求体尚未进入参数校验与依赖解析之前就需要被预处理。FastAPI 的请求参数（例如Body()声明的 JSON 体）最终都由APIRoute生成的处理器统一读取，因此只要把自定义逻辑挂在这条链路上，就能在一个位置覆盖所有路径操作，而不是在每个端点里重复书写。

需要认识的两个底层概念

在动手写自定义类之前，先厘清两个直接参与构造Request的 ASGI 概念（它们也出现在原文档的“技术细节（Technical Details）”提示框中）：

request.scope：一个承载请求元数据的 Pythondict（方法、路径、headers 等），属于 ASGI 规范的一部分；
request.receive：一个用于“接收”请求体数据的异步函数，同样源自 ASGI 规范。

关键在于：只要同时持有scope与receive，就可以构造出一个全新的Request实例。这是下面所有技巧的地基——GzipRequest(request.scope, request.receive)之所以可行，正源于此。关于Request更完整的方法与属性说明，可查阅 Starlette 的 Requests 文档（外部资料仅作参考，仓库内则直接阅读fastapi/requests.py与继承自 Starlette 的Request实现）。

同时需要理解APIRoute在请求生命周期中的角色。在 fastapi/routing.py 中可以看到：

APIRoute.__init__末尾会执行self.app = request_response(self.get_route_handler())（第 1223 行），即每个路由的 ASGI 应用本身就是get_route_handler()的返回值；
get_route_handler()（定义于第 1225 行）返回一个Callable[[Request], Coroutine[Any, Any, Response]]，内部通过get_request_handler(...)（第 1232 行）生成串联了依赖解析、参数校验、请求体读取、序列化响应的完整处理器。

由此可以推断：覆盖get_route_handler()等价于在“原始处理链”外面套一层自己的逻辑，从而实现对入站Request的替换与对出站Response的观察。

示例一：自定义 GzipRequest 与 GzipRoute 解压请求体

首先实现一个GzipRequest子类，覆盖Request.body()：当请求头带有合适的编码标识时解压请求体；当头部没有gzip时则不做解压尝试。这样同一个路由类可以同时正确处理 gzip 压缩与未压缩的请求。以下完整代码取自 docs_src/custom_request_and_route/tutorial001_an_py310.py：

import gzip from collections.abc import Callable from typing import Annotated from fastapi import Body, FastAPI, Request, Response from fastapi.routing import APIRoute 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 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 = FastAPI() app.router.route_class = GzipRoute @app.post("/sum") async def sum_numbers(numbers: Annotated[list[int], Body()]): return {"sum": sum(numbers)}

代码要点拆解：

惰性解压与结果缓存：body()首先检查实例上是否已有_body属性。gzip的 Python 实现是同步的，但这里解压发生在await super().body()拿到原始字节之后；用hasattr判断避免同一请求体被重复解压，同时把结果写回self._body，后续读取直接命中缓存。
头部判断：通过self.headers.getlist("Content-Encoding")取到可能存在的多个编码值，只有包含gzip时才调用gzip.decompress。缺失该头部时原样返回 body，因此不破坏普通请求。
路由处理器替换请求对象：GzipRoute.get_route_handler()先调用super().get_route_handler()拿到原始处理器，再包装出一个custom_route_handler：用GzipRequest(request.scope, request.receive)把收到的Request原地升级为GzipRequest，随后转交原始处理器继续执行。
全局接入：app.router.route_class = GzipRoute把根路由器的路由类整体替换。此后该应用下所有路径操作都会先经过GzipRequest的解压逻辑——当 FastAPI 因解析Body()而加载请求体时，获取到的已是解压后的数据，后续所有处理逻辑与普通请求完全一致，无需任何业务改动。

仓库同时提供了不使用Annotated的等价版本 tutorial001_py310.py（改用list[int] = Body()）。两个版本都需要 Python 3.10+，测试中通过needs_py310标记加以约束。

示例二：在异常处理器中访问请求体

相同思路也可以用于异常处理器。做法非常简单：在try/except块中处理请求，异常发生时Request实例仍然在作用域内，因此可以在处理错误时读取并利用请求体。完整代码见 docs_src/custom_request_and_route/tutorial002_an_py310.py：

from collections.abc import Callable from typing import Annotated from fastapi import Body, FastAPI, HTTPException, Request, Response from fastapi.exceptions import RequestValidationError from fastapi.routing import APIRoute 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)

custom_route_handler拦截了校验期抛出的RequestValidationError：通过exc.errors()拿到结构化的校验错误明细，通过await request.body()把原始请求体读出并decode()成字符串，一并放入HTTPException(status_code=422)的detail里。这样调用方在收到 422 时，能同时看到“错在哪里”和“你发来了什么”，极大方便排查。

💡 原文档特别指出：若只是想解决“在RequestValidationError自定义处理器里拿请求体”这一个问题，更简单的做法是直接在处理器中读取异常的body属性（见处理错误中关于RequestValidationError的章节）。本示例的价值在于演示如何与框架内部组件交互，它仍然完全有效。

示例三：通过 route_class 参数按路由定制 APIRoute

自定义路由类并不一定要作用于整个应用。APIRouter提供了route_class参数，可以精确控制该路由之下的所有路径操作。完整代码见 docs_src/custom_request_and_route/tutorial003_py310.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)

这个例子的巧妙之处在于它演示了路由类的作用域粒度：

挂载在TimedRoute之下的/timed端点：由于包装函数在original_route_handler(request)返回后才执行，它可以在响应已经生成但尚未发回客户端之前读取并修改response.headers——此处写入X-Response-Time头，值为生成响应所花费的秒数，同时把耗时、响应对象与响应头打印出来便于调试；
直接注册在应用上的/端点（not_timed）使用默认路由类，不会被计时包装，响应中自然也就没有X-Response-Time头。

从源码看 route_class 如何被消费

上述三个示例能成立，依赖的是 fastapi/routing.py 中APIRoute与APIRouter之间的既定协作：

APIRouter.__init__接收route_class参数（类型注解为type[APIRoute]，见第 2404 行附近），并保存为self.route_class（第 2562 行）；
在add_api_route（第 2889 行起）中，路由类按route_class = route_class_override or self.route_class解析（第 2921 行），即单条路由的route_class覆盖参数优先，否则回退到路由器的route_class；
随后以route = route_class(...)（第 2939 行）实例化具体路由。FastAPI实例的app.router就是一个普通的APIRouter，因此app.router.route_class = XxxRoute能对全局生效，而APIRouter(route_class=...)只能作用于该 router 及其 include 出去的路由。

结合第 1223–1249 行的实现可以确认完整的调用链：路由实例化时通过get_route_handler()生成处理函数 → 该函数被包装为 ASGI 应用 → 请求进入时调用它 → 子类覆盖的包装逻辑先于（或后于）original_route_handler执行。所有依赖解析、参数校验、请求体读取与响应序列化依旧由 FastAPI 原生完成，自定义类只需关注“请求进来之前 / 响应出去之前”这一小段窗口。

仓库测试对示例行为的验证

仓库在 tests/test_tutorial/test_custom_request_and_route/ 下为每个示例配备了自动化测试，可作为行为契约来对照：

test_tutorial001.py：通过parametrize("compress", [True, False])分别用 gzip 压缩与不压缩两种方式发送 1000 个整数的 JSON body，断言/sum均返回正确求和；同时新增/check-class探测端点断言请求对象类型名称为GzipRequest——这从侧面证实进入路径操作函数时，Request已被成功替换为自定义子类；
test_tutorial002.py：先验证合法 body[1, 2, 3]返回6；再发送{"numbers": [1, 2, 3]}这种结构错误的数据，断言 422 响应中的detail同时包含结构化的errors与原始 body 文本（测试注释还提示 httpx 0.28.0 起 JSON 可能以紧凑格式序列化，因此用IsOneOf兼容两种结果）；
test_tutorial003.py：断言/响应没有X-Response-Time头，而/timed响应有该头且其值可转成非负浮点数——精确验证了route_class的按路由器隔离效果。

需要说明的是，上述教程代码文件名中的py310表示其依赖list[int]、Annotated等 Python 3.10 语法；对应_an_变体为使用Annotated的推荐写法。

与中间件、异常处理器的取舍

原文档开篇即强调“这是对中间件逻辑的一种良好替代”，并在示例一中提示“若确实需要 gzip 支持，可直接使用框架自带的GzipMiddleware”（见 advanced/middleware.md）。综合三个示例，可给出如下取舍建议：

自定义APIRoute/Request：聚焦“某个/某组路由的请求体与响应”，能精确控制作用域（全局app.router.route_class或按APIRouter(route_class=...)），代码即路由配置的一部分，易于定位与测试；缺点是需要对框架内部对象有足够理解，属于进阶特性；
中间件：工作在更低层的 ASGI 层面，适合跨越所有路由、不关心路由语义的横切逻辑（如统一压缩、限流、日志），但对于“某个路由组”粒度的控制不如route_class直接；
异常处理器：如果只想在RequestValidationError时携带请求体返回错误信息，优先按处理错误中介绍的方式在自定义异常处理器里读取body属性，代码量最少。

企业官网建设流程全解析