FastAPI 自定义 Request 与 APIRoute 类：gzip 解压、异常中读取请求体与 route_class 路由定制实战

本文围绕 FastAPI 官方文档中「自定义 Request 和 APIRoute 类」这一高级主题展开，讲解如何通过继承 Request 重写 body() 方法、继承 APIRoute 重写 get_route_handler() 方法，实现 gzip 请求体自动解压、在异常处理器中访问原始请求体，以及通过 APIRouter 的 route_class 参数为整组路由注入自定义处理逻辑（如记录响应耗时）。读完本文，你将掌握 FastAPI 路由处理链的可扩展点、ASGI scope/receive 的协作方式，以及官方源码中该机制的真实实现路径。

这是一项“高级”功能：如果你刚开始学习 FastAPI，可以先跳过本篇，待熟悉了中间件、异常处理等基础机制后再回来阅读。自定义 Request 与 APIRoute 类在很多场景下是中间件的一种更聚焦的替代方案，尤其适合在请求体被应用真正处理之前，对其读取或进行转换操作的场景。

典型使用场景

官方文档列出的常见用例包括：

• 将非 JSON 编码的请求体（例如 msgpack 格式）转换为 JSON；
• 解压 gzip 压缩的请求体；
• 自动记录（log）所有请求体内容。

这些场景的共同特征是：逻辑必须发生在“FastAPI 解析请求体并交给路径操作函数”这一步之前，而恰好可以通过重写 Request.body() 或包装路由处理器来完成。

核心机制：Request 与 APIRoute 的可扩展点

在深入示例之前，先理解两个被扩展的类及其关键方法，这是本篇所有代码的基础：

• fastapi.Request（继承自 Starlette）：每次请求对应的对象。其 body() 方法负责读取并缓存请求体。FastAPI 在请求处理器内部正是通过 await request.body() 拿到请求体字节的，见 get_request_handler 中的请求体读取逻辑。因此重写 body() 就能透明地改变请求体在进入应用逻辑之前的形态。
• fastapi.routing.APIRoute：每个路径操作（Path Operation）对应的路由对象。它的关键方法是 get_route_handler()——该方法返回一个“接收 Request、返回 Response”的异步函数，由这个函数完成参数解析、依赖注入、响应校验等全部处理。从源码可以看到，路由初始化时即把它装配进 ASGI 应用：APIRoute.init 中 self.app = request_response(self.get_route_handler())，而 get_route_handler() 本身返回由模块级函数 get_request_handler(...) 构建的处理器。

这意味着：只要你的 APIRoute 子类重写了 get_route_handler()，就可以把原始处理器包一层，在调用前后做任何事情——换 Request 类型、计时、捕获异常等。

场景一：gzip 请求体的透明解压

以解压 gzip 压缩的请求体为例，展示完整的“自定义 Request + 自定义 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)}

自定义 GzipRequest 类：重写 body()

GzipRequest 只重写了一个方法 body()：

1. 先检查 _body 属性——这是 Starlette Request 缓存请求体的惯用属性，避免重复读取（请求体流通常只能消费一次）；
2. 调用 super().body() 取得原始字节；
3. 仅当请求头 Content-Encoding 的列表中包含 gzip（self.headers.getlist("Content-Encoding") 会返回该头所有值组成的列表）时，才用 gzip.decompress(body) 解压；
4. 将结果缓存到 self._body，后续调用直接复用。

这个设计的一个优点是：没有 Content-Encoding: gzip 头的普通请求会原样通过，因此同一路由可以同时接受 gzip 压缩与未压缩的请求。

提示：上面是一个演示原理的简单示例。如果你在生产环境只需要 gzip 支持，官方建议直接使用现成的 GzipMiddleware，参见德语文档 GzipMiddleware 说明。

自定义 GzipRoute 类：把普通 Request 换成 GzipRequest

GzipRoute 继承 fastapi.routing.APIRoute，重写 get_route_handler()。该方法的约定是：返回一个异步函数，函数接收一个 Request 并返回一个 Response。这里我们用它做一件事——在把请求交给原始处理器之前，用原请求的 scope 和 receive 重新构造一个 GzipRequest：

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

技术细节（为什么是 scope + receive）：每个 Request 对象都有一个 request.scope 属性，它就是一个普通的 Python dict，保存与该请求关联的所有元数据；同时还有一个 request.receive 属性，它是一个用于“接收”请求体数据的可调用函数。scope 字典与 receive 函数这两者都是 ASGI 规范的组成部分，也是构造新 Request 实例所必需的两个参数。GzipRequest(request.scope, request.receive) 正是基于这一点，把同一个底层 ASGI 请求“换皮”成自定义类型。

GzipRoute.get_route_handler 返回的函数与默认实现的唯一区别，就是把 Request 转换成了 GzipRequest。由此，我们的 GzipRequest 会在数据交给路径操作函数之前按需完成解压；之后所有的处理逻辑（参数解析、校验、依赖、响应处理）完全不变。由于改的是 GzipRequest.body，当 FastAPI 内部读取请求体时，字节流会在需要时被自动解压。

启用方式：示例中通过 app.router.route_class = GzipRoute 直接修改了 FastAPI 主路由器的 route_class，使应用中所有路由都走 GzipRoute。也可以只对部分路由生效（见下文“自定义 APIRoute 用于 Router”一节）。

场景二：在异常处理器中访问原始请求体

同样的“包装路由处理器”技巧，也可以用来在异常处理时读取请求体。官方同时给出一个提示：如果目标只是在 RequestValidationError 中拿到 body，通常更简单的做法是直接自定义该异常的处理器并在其中使用 exc.body，参见德语文档 错误处理章节的相应说明。本示例的价值在于演示如何与 FastAPI 内部组件交互。

完整代码见 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)

要点拆解：

• 唯一的新逻辑是把 original_route_handler(request) 放进 try/except 块中：

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)

• 当请求体验证失败时抛出 RequestValidationError，此时 Request 实例仍在作用域内，所以可以 await request.body() 读到原始字节，把它（连同 exc.errors() 的错误明细）一起放进 422 响应的 detail 中返回给客户端。
• app.router.route_class = ValidationErrorLoggingRoute 让应用内所有路由都具备这个“带请求体回显”的 422 行为。

这个模式可以推广：任何“需要在路径操作执行前后插入逻辑，并可能捕获特定异常”的需求，都可以照此包装 get_route_handler() 的返回值。

场景三：通过 route_class 参数让 Router 使用自定义 APIRoute

除了直接改写 app.router.route_class，官方推荐的方式是给 APIRouter 传 route_class 参数，实现按路由器粒度启用自定义路由类。完整代码见 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)

在这个示例中，挂在 router 下的路径操作都会使用自定义的 TimedRoute 类：TimedRoute 在响应返回前计算处理耗时，并写入响应头 X-Response-Time（同时打印耗时、响应对象与响应头用于演示）。而直接挂在 app 上的 GET / 路由不受影响，不会带上这个头。

从源码可以印证这一机制的完整链路：

• APIRouter.__init__ 接收 route_class 参数（默认值 APIRoute），文档字符串明确写着“Custom route (path operation) class to be used by this router”，并保存到 self.route_class，见 APIRouter 的 route_class 参数定义 与 赋值语句；
• 当通过装饰器向该路由器添加路径操作时，FastAPI 会用 route_class_override or self.route_class 选出实际的路由类，并 route = route_class(...) 实例化，见 add_api_route 中的路由类选择。

也就是说，route_class 决定了“该路由器内每一条路径操作具体实例化成哪个 Route 类”，而每个 Route 实例在初始化时又会调用自己的 get_route_handler() 生成处理器（见前文 APIRoute.init），两者结合就完成了整条链路。

源码与测试佐证

• 路由处理器装配点：fastapi/routing.py——APIRoute.__init__ 第 1223 行 self.app = request_response(self.get_route_handler())，get_route_handler() 委托给模块级函数 get_request_handler()。
• 请求体读取点：fastapi/routing.py——非表单请求下执行 body_bytes = await request.body() 再解析 JSON。这解释了为什么重写 Request.body() 能透明影响后续全部处理。
• route_class 机制：fastapi/routing.py（参数定义）、fastapi/routing.py（add_api_route 中选用路由类并实例化）。
• 行为测试：tests/test_custom_route_class.py 用三个不同的自定义 APIRoute 子类（APIRouteA/APIRouteB/APIRouteC）分别构造 APIRouter(route_class=...)，验证每个路由器内的路径操作确实使用了各自的路由类。
• 请求/响应校验失败的默认行为可结合 RequestValidationError 定义 理解；ValidationErrorLoggingRoute 示例即是在其抛出位置做了拦截与增强。

小结与选型建议

| 需求 | 推荐做法 | | |---| | 处理 gzip 请求体 | 优先用现成的 GzipMiddleware；需要更细粒度控制时参考 GzipRequest + GzipRoute 模式 | | 在 422 错误中回显请求体 | 优先自定义 RequestValidationError 异常处理器（直接用 exc.body）；需要更深层拦截时用 route_class 包装 get_route_handler() | | 给某组路由统一加计时/审计/响应头 | 自定义 APIRoute 子类 + APIRouter(route_class=...)，只影响该路由器的路径操作 | | 全局替换路由行为 | 设置 app.router.route_class = YourRoute |

两个核心记忆点：

1. 改请求体输入：继承 Request，重写 body()，并注意用 _body 做一次性缓存；
2. 改处理流程：继承 APIRoute，重写 get_route_handler()，用“先取 super().get_route_handler()，再包装一层”的方式注入逻辑。

注意该机制属于高级用法：包装函数中若抛出非预期异常会绕过默认的 422/500 行为，且 route_class 只作用于该路由器后续通过装饰器/add_api_route 添加的路径操作。修改前建议先用 TestClient 覆盖正常、压缩、非法请求体三类请求，确认行为符合预期。