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属性它只是一个包含请求元数据的 PythondictRequest还有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.pyclass 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)要点只需把对原始路由处理器的调用包进try/except块发生异常时Request实例仍然在作用域内因此可以await request.body()读取并复用已缓存的请求体将其与校验错误一起放进 422 响应的detail中。文档同时提醒如果只是想在校验失败时拿到请求体更简单的方式是在RequestValidationError的自定义异常处理器里直接使用异常对象自带的body属性参见 Handling Errors 文档。本例的价值在于演示如何与框架内部组件交互这一通用模式。实战三为某个APIRouter单独指定route_class自定义路由类不必全局生效APIRouter提供了route_class参数源码中默认值就是APIRoute见 fastapi/routing.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)完整示例见 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 的实现可以把这套机制串成一条清晰的调用链路由初始化时绑定处理器APIRoute.__init__结束时执行self.app request_response(self.get_route_handler())第 1223 行。因为 Python 的多态机制如果你的子类重写了get_route_handler()这里拿到的就是子类版本——这就是为什么只覆写一个方法就能改变整个路由行为。get_route_handler()返回真正的请求处理器默认实现第 1225-1249 行调用get_request_handler(...)把路由的依赖dependant、body 字段、响应模型等配置打包成一个async def app(request) - Response协程函数负责读取 body、解析依赖、执行端点函数、序列化响应。子类通过装饰原始处理器介入文档中所有例子都遵循同一模式——先original_route_handler super().get_route_handler()拿到原始处理器再返回一个闭包custom_route_handler在调用前后插入自定义逻辑换请求对象、包 try/except、加响应头。这种包裹模式保证了你无需理解框架内部细节只需围绕原始处理器加一层。请求体读取是唯一的耦合点get_request_handler生成的处理器中body 的读取是await request.body()第 433 行和await request.form()第 430 行。因此GzipRequest只需覆写body()就能让后续所有 JSON 解析、依赖注入透明地拿到解压后的数据。应用级启用方式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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 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/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属性它只是一个包含请求元数据的 PythondictRequest还有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.pyclass 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)要点只需把对原始路由处理器的调用包进try/except块发生异常时Request实例仍然在作用域内因此可以await request.body()读取并复用已缓存的请求体将其与校验错误一起放进 422 响应的detail中。文档同时提醒如果只是想在校验失败时拿到请求体更简单的方式是在RequestValidationError的自定义异常处理器里直接使用异常对象自带的body属性参见 Handling Errors 文档。本例的价值在于演示如何与框架内部组件交互这一通用模式。实战三为某个APIRouter单独指定route_class自定义路由类不必全局生效APIRouter提供了route_class参数源码中默认值就是APIRoute见 fastapi/routing.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)完整示例见 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 的实现可以把这套机制串成一条清晰的调用链路由初始化时绑定处理器APIRoute.__init__结束时执行self.app request_response(self.get_route_handler())第 1223 行。因为 Python 的多态机制如果你的子类重写了get_route_handler()这里拿到的就是子类版本——这就是为什么只覆写一个方法就能改变整个路由行为。get_route_handler()返回真正的请求处理器默认实现第 1225-1249 行调用get_request_handler(...)把路由的依赖dependant、body 字段、响应模型等配置打包成一个async def app(request) - Response协程函数负责读取 body、解析依赖、执行端点函数、序列化响应。子类通过装饰原始处理器介入文档中所有例子都遵循同一模式——先original_route_handler super().get_route_handler()拿到原始处理器再返回一个闭包custom_route_handler在调用前后插入自定义逻辑换请求对象、包 try/except、加响应头。这种包裹模式保证了你无需理解框架内部细节只需围绕原始处理器加一层。请求体读取是唯一的耦合点get_request_handler生成的处理器中body 的读取是await request.body()第 433 行和await request.form()第 430 行。因此GzipRequest只需覆写body()就能让后续所有 JSON 解析、依赖注入透明地拿到解压后的数据。应用级启用方式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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考