FastAPI 自定义 Request 与 APIRoute 类从 gzip 解压到请求体审计的完整实战【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi自定义Request与APIRoute类是 FastAPI 中一类「进阶」但极具威力的扩展机制它允许你改写请求在进入路径操作path operation之前的处理逻辑从而在不引入全局中间件的前提下完成请求体解压、格式转换、自动日志、异常诊断与响应计时等横切需求。读完本文你将掌握三个可落地的方案用GzipRequest透明解压 gzip 请求体、在异常处理器中读取原始请求体、以及通过route_class为单个路由或整组路由注入计时与审计能力并理解其背后get_route_handler()与 ASGIscope/receive的调用原理。本文对应的官方文档为 docs/pt/docs/how-to/custom-request-and-route.md英文版见 docs/en/docs/how-to/custom-request-and-route.md全部示例代码位于 docs_src/custom_request_and_route/ 目录。⚠️ 进阶提示这是一个「高级」特性。如果你是 FastAPI 初学者可以先跳过本节待熟悉路径操作、依赖注入与中间件之后再回来阅读。为什么需要自定义 Request 与 APIRouteFastAPI 的每个路径操作背后都有一个对应的APIRoute实例负责解析请求、校验参数、调用端点函数、封装响应。而这个流程的最外层是一个接收Request、返回Response的异步处理函数——它正是通过APIRoute.get_route_handler()生成的。在有些场景下你希望在请求体被应用处理之前读取或改写它此时自定义Request/APIRoute是比中间件middleware更精准的替代方案。官方文档列出的典型用例包括将非 JSON 的请求体转换为 JSON例如msgpack编码解压 gzip 压缩的请求体自动记录logging所有请求体内容。与全局中间件相比这种做法的优势是作用域可控你可以把自定义行为精确地绑定到某个路由、某个APIRouter或整个应用而非所有请求。场景一处理自定义请求体编码——gzip 解压下面以「透明解压 gzip 请求体」为例展示完整的自定义链路。完整代码见 docs_src/custom_request_and_route/tutorial001_an_py310.py。 说明这是一个用于演示机制原理的 toy example。如果你只是需要 gzip 支持可以直接使用 FastAPI 提供的GzipMiddleware针对响应压缩而本文的自定义Request处理的是请求体的解压方向两者解决的问题不同。第一步创建自定义GzipRequest类自定义的入口是重写Request.body()方法当请求头Content-Encoding中包含gzip时把读取到的原始字节用gzip.decompress()解压否则原样返回。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几个值得注意的实现细节用self._body做了结果缓存body()被多次调用时例如依赖注入与端点函数各读一次不会重复解压用self.headers.getlist(Content-Encoding)而不是self.headers.get(Content-Encoding)因为 HTTP 允许该头出现多个值如Content-Encoding: gzip, brgetlist能正确处理逗号分隔列表如果请求头中没有gzip则完全跳过解压逻辑因此同一个路由类既能处理压缩请求也能处理未压缩请求。第二步创建自定义GzipRoute类有了请求类还不够还需要让它真正被路由使用。做法是继承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()返回一个「接收Request、返回Response」的异步函数它就是 ASGI 层与路径操作之间的桥。这里唯一做的新事情是用原始请求的scope与receive构造一个新的GzipRequest实例其余处理逻辑全部委托给original_route_handler原样执行。技术细节scope与receive是什么这里涉及 ASGI 规范的两个核心概念request.scope一个 Pythondict包含与当前请求相关的元数据请求方法、路径、查询参数、请求头、客户端地址等request.receive一个可调用对象用于异步「接收」请求体数据。这两者正是创建新Request实例所需的全部输入——Request(scope, receive)。因为GzipRequest只是把body()行为换掉了scope/receive可以原样透传所以 ASGI 层的握手、头信息解析等逻辑完全不受影响。在 FastAPI 中Request直接复用 Starlette 的实现见 fastapi/requests.pyfrom starlette.requests import Request as Request。第三步让应用使用自定义路由类app FastAPI() app.router.route_class GzipRoute app.post(/sum) async def sum_numbers(numbers: Annotated[list[int], Body()]): return {sum: sum(numbers)}将app.router.route_class赋值为GzipRoute后应用上注册的所有路径操作都会改用这个路由类。此时向/sum发送Content-Encoding: gzip的压缩 JSON 请求体FastAPI 加载请求体时会自动调用被重写的body()从而透明完成解压端点函数拿到的就是解压后的原始 JSON。场景二在异常处理器中访问请求体同一个套路也可以用来「抢救」请求体当请求校验失败抛出RequestValidationError时在自定义路由处理函数中用try/except捕获异常并读取request.body()把原始请求体一并写入错误详情。完整代码见 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_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)关键点在于异常发生时Request实例仍然在custom_route_handler的作用域内因此可以读取并利用请求体来辅助诊断错误。这里把校验错误列表exc.errors()与原始请求体文本一起放进 422 响应的detail方便前端或日志系统定位问题。 提示如果只是为了解决「校验失败时返回请求体」这一个问题更简单的做法是在自定义RequestValidationError处理器中读取request.body()参见 docs/pt/docs/tutorial/handling-errors.md 中关于RequestValidationError的部分。但本文这个示例仍然有效并且展示了如何与内部组件交互——当你需要在这类钩子里做更复杂的逻辑时这套模式可以直接迁移。场景三在 APIRouter 级别注入自定义路由类自定义行为不必总是全局生效。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(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在调用原始处理函数前后用time.time()计时并把耗时写入响应头X-Response-Timerouter APIRouter(route_classTimedRoute)之后只有挂载在该router下的路径操作/timed会带上计时逻辑直接定义在app上的/路由不受影响——这正是「按路由分组定制」相对全局中间件的灵活性所在。源码级原理route_class与get_route_handler()的调用链理解了三个实战场景后再回到源码确认底层机制能帮你更自信地扩展这套体系。相关实现集中在 fastapi/routing.py。APIRoute.get_route_handler()请求处理函数的工厂APIRoute的__init__中有一行self.app request_response(self.get_route_handler())fastapi/routing.py即每个路由的 ASGI 应用正是由get_route_handler()产出的。默认实现fastapi/routing.py会基于该路由的依赖图dependant、请求体字段body_field、状态码、响应类等元数据调用get_request_handler()生成最终的处理函数。因此当你重写get_route_handler()并调用super().get_route_handler()时拿到的是 FastAPI 为你组装好的完整处理管线你只需在其外层包裹自定义逻辑换请求类、捕获异常、计时、记录日志即可在不破坏任何既有校验、依赖注入与序列化能力的前提下注入横切行为。route_class路由类的注入点APIRouter构造函数的route_class参数声明为type[APIRoute]默认值是APIRoute本身fastapi/routing.py。在api_route/add_api_route内部路由实例化时遵循「显式覆盖优先」的规则route_class route_class_override or self.route_class ... route route_class( self.prefix path, endpointendpoint, response_modelresponse_model, ... )见 fastapi/routing.py 与 fastapi/routing.py。这意味着设置app.router.route_class GzipRoute后FastAPI应用顶层路由器创建的所有路由都会使用GzipRoute单个APIRouter(route_classTimedRoute)则把作用域限定在该 router 内部如果需要更细的控制还可以利用route_class_override机制在单个路径操作级别覆盖从源码结构看这一参数由内部调用链传入一般情况下你直接使用前两种方式即可。Request的来源FastAPI 的Request是对 StarletteRequest的直接复用见 fastapi/requests.py因此你自定义的子类天然继承 Starlette 的全部能力scope、receive、headers、path_params、query_params、cookies、client等属性与方法均可直接使用。测试验证机制确实生效仓库为这三个示例都编写了自动化测试位于 tests/test_tutorial/test_custom_request_and_route/可作为「机制确实生效」的实证。以 tests/test_tutorial/test_custom_request_and_route/test_tutorial001.py 为例test_gzip_request使用TestClient分别以压缩gzip.compressContent-Encoding: gzip和未压缩两种方式向/sum提交[1] * 1000的 JSON 请求体两次都断言返回{sum: 1000}——证明同一个路由类能同时处理压缩与未压缩请求test_request_class额外注册了一个检查路由断言请求实例的类名是GzipRequest——证明路由确实把原始Request替换成了自定义子类。其余两个测试文件分别覆盖异常日志路由tutorial002与计时路由tutorial003运行pytest tests/test_tutorial/test_custom_request_and_route/即可验证全部行为。小结自定义Request与APIRoute是 FastAPI 为「请求进入路径操作之前」这一时机提供的精细化扩展点与中间件形成互补能力中间件自定义Request/APIRoute作用域全局或挂载点下的全部请求可按路由 / 路由组 / 应用精确控制读取请求体需自行消费 ASGI 消息直接重写body()透明生效包裹端点执行不直接感知端点可精确包裹处理函数计时、异常捕获复杂度较低较高中进阶特性当你需要「在应用处理之前读取或改写请求体」且希望行为精准落在某个路由子集上时GzipRequest、ValidationErrorLoggingRoute、TimedRoute这三套模式分别对应请求体改写、异常诊断、响应审计足以覆盖绝大多数需求。继续深入可研读 fastapi/routing.py 中APIRoute/APIRouter的完整实现或参考官方文档 docs/en/docs/how-to/custom-request-and-route.md 获取最新说明。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考