FastAPI 自定义 Request 与 APIRoute 深入实战改写请求体、在异常处理器读取 Body 与路由级耗时统计【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi在 FastAPI 中Request与APIRoute是每个请求进入path operation前最核心的两个组件前者承载 HTTP 请求的 scope 元数据与 body后者负责把请求转交给具体的处理函数。多数情况下开发者不需要触碰它们但当需要在请求进入业务逻辑之前统一读取、转换或记录 body且希望把这种逻辑以请求级/路由级的粒度而非全局中间件的方式实现时FastAPI 官方 How-To 指南 custom-request-and-route 提供了一个通用且可扩展的进阶方案。读完本文你将掌握如何通过继承Request覆写body()、如何通过继承fastapi.routing.APIRoute覆写get_route_handler()以及如何在APIRouter层面按路由组注入自定义行为——并用仓库中的源码与测试验证每个环节的真实工作机制。什么时候需要自定义 Request 与 APIRoute⚠️ 这是一项advanced进阶特性。如果你刚开始接触 FastAPI可以暂时跳过本主题先掌握常规的请求处理与中间件用法。覆写Request和APIRoute的逻辑本质上是中间件之外的一种更精细的替代方案它更贴近单个路由的请求生命周期。官方文档给出的典型 use cases 包括将非 JSON 请求体转换为 JSON例如接收msgpack这类二进制编码的 body在交给 FastAPI 解析 JSON 之前先解码/反序列化解压 gzip 压缩的请求体客户端用Content-Encoding: gzip压缩提交服务端在读取 body 时先解压自动记录所有请求体在 body 进入应用处理前统一读取并落日志。其共同特点是对 body 的事前介入——恰好是纯 ASGI 中间件很难优雅处理的部分因为 body 是流式的一旦被中间件读取后续链路就难以再取用原始内容。而自定义Request.body()则能把转换逻辑内置在读取动作本身。官方在文中特别强调这更多是演示机制的教学示例。比如真的需要 Gzip 支持优先使用 FastAPI 自带的GzipMiddleware见 advanced/middleware.md 中的 GzipMiddleware 小节本文只是借助这类场景说明如何解开内部组件。第一招自定义GzipRequest覆写Request.body()官方完整示例位于 docs_src/custom_request_and_route/tutorial001_an_py310.py。第一步是定义一个继承自Request的子类覆写其body()方法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这里的关键逻辑有三点仅在存在Content-Encoding: gzip时才解压。Request.headers.getlist(Content-Encoding)以列表形式返回该 header 的全部取值避免多个编码值如gzip, deflate时漏判不做解压的请求则原样透传因此同一条路由可以同时服务压缩与非压缩的客户端用self._body做结果缓存后续任何代码再次调用body()时不会重复读取流或重复解压。技术细节scope与receive是什么覆写过程中有一个概念必须澄清。一个Request对象本质上由两样东西驱动这也是 ASGI 规范定义的两大输入request.scope一个 Pythondict保存请求相关的元数据如 method、path、query string、headers 等request.receive一个异步函数用于从底层 receive 请求 body 的消息。scope与receive都来自 ASGI 规范构造一个新的Request实例只需要把这两者传入构造函数即可。FastAPI 的Request实现继承自 Starlette关于Request的更详细行为可参考 Starlette 官方 Requests 文档文中对应位置即为该 note 的出处。第二招自定义GzipRoute覆写APIRoute.get_route_handler()有了GzipRequest还需要让 FastAPI 真正使用它。这就要继承fastapi.routing.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_handlerget_route_handler()是 FastAPI 内部用来构造每个路由最终执行入口的方法它返回一个接收Request、返回Response的函数。覆写它的模式非常固定先调用super().get_route_handler()拿到原装的路由处理函数保留全部既有逻辑依赖注入、参数解析、校验、序列化等在外层包一个custom_route_handler在这里把收到的普通Request替换成我们的GzipRequest注意必须同时传入request.scope与request.receive两个构造必需参数把替换后的请求继续交给原 handler 处理其余逻辑一概不变。从源码理解get_route_handler的地位在 fastapi/routing.py 中APIRoute类的构造尾声有这样一行routing.py#L1223self.app request_response(self.get_route_handler())也就是说每个APIRoute实例在初始化时就把get_route_handler()的返回值包装成一个 ASGI 应用self.app后续 ASGI 服务器通过handle()调用这个 app 来执行请求。而原生get_route_handler()内部则是调用get_request_handler(...)把dependant、body_field、status_code、response_class、response_model_*等一整套参数打包成最终处理器routing.py#L1225-L1249。因此当我们 override 时只要调用super().get_route_handler()就能把这套完整的依赖解析与响应处理管线原样保留只在外层偷换请求对象。完成两个类后把它们接到应用上——FastAPI实例的router本身就是APIRouter所以可以直接改写其route_classapp FastAPI() app.router.route_class GzipRoute app.post(/sum) async def sum_numbers(numbers: Annotated[list[int], Body()]): return {sum: sum(numbers)}当客户端POST /sum并携带 gzip 压缩过的Content-Encoding: gzip请求体时body 会在 FastAPI 需要加载请求体做 JSON 解析的过程中被GzipRequest.body()自动解压于是list[int]校验与sum()业务逻辑拿到的都是解压后的干净数据开发者的 path operation 代码完全无感知。第三招在异常处理器中读取请求体同一个包装 route handler思路也能用来访问异常发生时的请求体。完整示例见 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)机制说明把对原 handler 的调用放进try/except RequestValidationError块中一旦校验失败抛出RequestValidationError此时request对象仍在闭包作用域内可立即调用await request.body()读取原始 body把exc.errors()结构化的校验错误列表与原始 body 文本一起包装成detail再抛出一个422 HTTPException从而把出错时到底收到了什么数据一并返回给客户端或记录到日志。需要提醒若只是想把校验失败的原始请求体附到响应里官方更推荐的做法是在RequestValidationError的全局自定义异常处理器中直接使用其自带的body属性参见 handling-errors.md 中 Use the RequestValidationError body 一节本文示例的价值在于演示如何与 FastAPI 的内部组件直接交互。测试如何验证这段逻辑仓库 tests/test_tutorial/test_custom_request_and_route/test_tutorial002.py 给出了行为级验证当POST /提交合法 JSON 数组[1, 2, 3]时返回求和结果6当提交{numbers: [1, 2, 3]}不符合list[int]时返回的 422 响应detail中同时包含类型为list_type的errors数组以及原文{numbers: [1, 2, 3]}的body字段——证明异常处理分支确实读到了原始请求体。第四招在APIRouter级别注入自定义APIRoute前三招都是全局替换app.router.route_class。如果只想让一部分路由启用自定义行为则应在创建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在调用完原 handler 之后用time.time()的差值算出处理耗时并把它写入响应的X-Response-Time自定义 header同时打印耗时、响应对象与其 headers。由于只有APIRouter(route_classTimedRoute)声明的router使用了该类挂在router下的/timed会返回带X-Response-Time头的响应直接挂在app上的/仍走默认APIRoute不带该 header。测试如何验证路由级隔离对应测试 tests/test_tutorial/test_custom_request_and_route/test_tutorial003.py 清晰地断言了这一点请求/时响应头中不存在X-Response-Time请求/timed时响应头中存在X-Response-Time且其数值 0。这正好印证route_class的作用范围是这个 router 内部的 path operations而非全局。从源码看route_class如何生效在 fastapi/routing.py 的APIRouter.__init__中route_class是一个注解为type[APIRoute]、默认值为APIRoute的构造参数routing.py#L2404-L2414并被保存在实例属性self.route_classrouting.py#L2562。随后在add_api_route()注册路由时route_class route_class_override or self.route_class ... route route_class(...) # 用选定的 route_class 构造 APIRoute 实例见 routing.py#L2921 与 routing.py#L2939。也就是说FastAPI()顶层装饰器最终调用的也是其内部router的add_api_route因此设置app.router.route_class等价于对应用内全部路由生效而给某个APIRouter单独传route_class则只影响 include 进来的那一组路由。三种机制的分工与取舍方案作用范围适合场景关键覆写点全局app.router.route_class XxxRoute应用内全部路由全站统一的 body 预处理 / 日志 / 编码转换APIRoute.get_route_handler()单个APIRouter(route_classXxxRoute)仅该 router 下的路由部分 API 分组启用计时、鉴权日志等同上作用面收窄自定义Request子类配合上述 route 使用必须改写 body 读取行为本身Request.body()使用中的几条实用建议覆写body()时务必做结果缓存如示例的self._body避免同一请求内多次读取时重复消费 ASGI 消息流尽量让同一路由兼容两种输入压缩与非压缩正如GzipRequest只在 header 含gzip时才解压避免误伤常规客户端get_route_handler()覆写必须调用super()否则会丢失 FastAPI 的依赖解析、参数校验与响应序列化整条管线若目标只是 gzip 压缩这类常见需求请优先选用GzipMiddleware自定义类是学习路由级逻辑注入的通用载体而非银弹。小结围绕在请求进入业务逻辑前对 Request/Route 做定制这一主题FastAPI 给出了完整的进阶链路继承Request改变 body 的读取语义继承fastapi.routing.APIRoute并覆写get_route_handler()来包装每个请求的处理过程再借助route_class将自定义路由类作用到全局或某个子路由上。仓库中的 docs_src/custom_request_and_route 示例、fastapi/routing.py 的实现self.app request_response(self.get_route_handler())与add_api_route中route_class(...)的调用以及 tests/test_tutorial/test_custom_request_and_route 下的三个测试用例共同构成了从机制到用法再到验证的完整闭环——理解了这一层你就能在需要中间件力所不能及的细粒度控制时自如地潜入 FastAPI 的请求管线。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考