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/fastapiFastAPI 在底层把每一个路径操作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.pyimport 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.pyfrom 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_code422, detaildetail) 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_code422)的detail里。这样调用方在收到 422 时能同时看到“错在哪里”和“你发来了什么”极大方便排查。 原文档特别指出若只是想解决“在RequestValidationError自定义处理器里拿请求体”这一个问题更简单的做法是直接在处理器中读取异常的body属性见 处理错误 中关于RequestValidationError的章节。本示例的价值在于演示如何与框架内部组件交互它仍然完全有效。示例三通过 route_class 参数按路由定制 APIRoute自定义路由类并不一定要作用于整个应用。APIRouter提供了route_class参数可以精确控制该路由之下的所有路径操作。完整代码见 docs_src/custom_request_and_route/tutorial003_py310.pyimport 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(froute duration: {duration}) print(froute response: {response}) print(froute response headers: {response.headers}) return response return custom_route_handler app FastAPI() router APIRouter(route_classTimedRoute) app.get(/) async def not_timed(): return {message: Not timed} router.get(/timed) async def timed(): return {message: Its 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属性代码量最少。延伸阅读本文对应英文原版custom-request-and-route.md韩文版即本次基准文档 docs/ko/docs/how-to/custom-request-and-route.md中文翻译版见 docs/zh/docs/how-to/custom-request-and-route.md教程示例源码目录docs_src/custom_request_and_route/含py310与_an_py310双版本路由核心实现与get_route_handler/route_class消费逻辑fastapi/routing.py对应行为验证测试tests/test_tutorial/test_custom_request_and_route/相关主题中间件与 GzipMiddleware、处理错误含 RequestValidationError body 用法【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考