在 FastAPI / Uvicorn 中接受 gzipped body

Question

我正在将FastAPI与Uvicorn一起使用来实现 u-service，它接受请求正文中的 json 有效负载。由于请求正文可能非常大，我希望服务接受 gzip 压缩。我该如何做到这一点？

到目前为止，如下：

• 添加了GZip 中间件，但它对响应进行编码，而不是解码请求
• 在我的请求中添加了“内容编码：gzip”

失败并响应：

状态：400 错误请求
{“detail”：“解析正文时出错”}

Answer 1

FastAPI文档包含自定义 gzip 编码请求类的示例。

注意：此页面还包含以下短语：“...如果您需要 Gzip 支持，可以使用提供的 GzipMiddleware。” ，但这是不正确的，因为您正确地注意到中间件仅适用于响应。

import gzip
from typing import Callable, List

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: List[int] = Body(...)):
    return {"sum": sum(numbers)}