Accept gzipped body in FastAPI / Uvicorn

Question

I'm using FastAPI with Uvicorn to implement a u-service which accepts a json payload in the request's body. Since the request body can be quite large, I wish the service to accept gzipped. How do I accomplish that?

So far the following:

• added the GZipMiddleware, but it encodes responses, rather that decoding requests
• added a 'Content-Encoding: gzip' to my request

Fail with response:

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

Answer 1

FastAPI documentation contains an example of a custom gzip encoding request class.

Note: This page also contains the following phrase: "...if you need Gzip support, you can use the provided GzipMiddleware.", but this is incorrect, since you correctly noticed that middleware only works for responses.

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

Another way to achieve the same would be like this:

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")