企业官网建设流程全解析

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

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

本文围绕 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属性，它只是一个包含请求元数据的 Pythondict；
• 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 的实现，可以把这套机制串成一条清晰的调用链：

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

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