在 FastAPI / Uvicorn 中接受 gzipped body

Question

我正在使用 FastAPI with Uvicorn 实现一个 u 服务，该服务接受请求正文中的 json 有效负载。由于请求正文可能非常大，我希望服务接受 gzip 压缩。我该如何实现？

到目前为止如下：

添加了 GZipMiddleware，但它编码响应，而不是解码请求
在我的请求中添加了 'Content-Encoding: gzip'

响应失败：

Status: 400 Bad Request
{ "detail": "There was an error parsing the body" }

Answer 1

FastAPI 文档 contains 自定义 gzip 编码请求的示例 class。

注意：该页面还包含以下短语："...如果您需要 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)}

Answer 2

实现相同目的的另一种方法是这样的：

from fastapi import FastAPI
from starlette.types import Message
from starlette.requests import Request
from starlette.middleware.base import BaseHTTPMiddleware
import gzip          
        

class GZipedMiddleware(BaseHTTPMiddleware):
    async def set_body(self, request: Request):
        receive_ = await request._receive()
        if "gzip" in request.headers.getlist("Content-Encoding"):
            print(receive_)                                     
            data = gzip.decompress(receive_.get('body'))
            receive_['body'] = data

        async def receive() -> Message:
            return receive_

        request._receive = receive                

    async def dispatch(self, request, call_next):
        await self.set_body(request)        
        response = await call_next(request)                
        return response

    

app = FastAPI()

app.add_middleware(GZipedMiddleware)


@app.post("/post")
async def post(req: Request):
    body = await req.body()
    # I'm decoding here in case you just gziped an string
    return body.decode("utf-8")

在 FastAPI / Uvicorn 中接受 gzipped body

Accept gzipped body in FastAPI / Uvicorn

python

fastapi

uvicorn