FastAPI自学教程（86） - 自定义请求与路由类

1.基础用法

import gzip
from fastapi import 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

• 通过继承Request类实现Gzip请求体自动解压缩
• 使用自定义APIRoute类覆盖路由处理逻辑，支持特殊编码处理
• 处理流程：

  graph TD
    A[客户端请求] --> B{Content-Encoding头}
    B -->|包含gzip| C[GzipRequest解压]
    B -->|无压缩| D[原始请求处理]
    C/D --> E[路由处理函数]
    E --> F[返回响应]

2.核心配置

2.1 异常处理集成

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()
                logger.error(f"请求体解析错误: {body}")
                raise exc
        return custom_route_handler

• 在异常处理中捕获原始请求体，实现调试日志记录
• 支持在HTTPException处理前访问原始请求数据

2.2 多阶段请求处理

• 请求体预解析：在FastAPI默认解析前处理原始字节流
• 动态编码支持：根据请求头自动选择解码器（如msgpack/protobuf）
• 数据转换：将非JSON格式转换为标准数据结构

3.生产环境建议

1. 安全加固：

   • 限制解压后数据大小（防止内存溢出攻击）
   • 校验Content-Encoding头合法性

2. 性能优化：

   app = FastAPI(debug=False)  # 关闭调试模式

   • 使用lru_cache缓存解码器实例
   • 对高频请求禁用复杂解码

3. 监控体系：

   • 记录异常请求的原始字节流用于审计
   • 统计各编码格式请求的耗时指标

4. 版本管理：

   • 通过路由类继承实现多版本API共存
   • 使用装饰器标注支持的编码格式版本

完整实现参考：
FastAPI官方文档 - 自定义请求与路由类
测试命令示例：

# 发送Gzip压缩请求
echo '[1,2,3]' | gzip | curl -X POST http://localhost:8000/sum \
  -H "Content-Encoding: gzip" \
  --data-binary @-