Originally published on my blog

Now, as I promised on the past article I'm going to build a more complicated middleware. Most of the job will be made by a library called google/brotli, given that I'm not interested on making an implementation of the Brotli algorithm. Instead of that, my main pupose here, is to be able to implement a middleware that behave like the GZipMiddleware, from the FastAPI documentation.

About the Brotli algorithm.

As you may imagine I'm not an expert on compression algorithm, so anything I'm going to say is a result of browsing the web.

Quoting MDN one of the most reliable cites about web development:

Brotli provides better compression ratios than gzip and deflate speeds are comparable, but brotli compressing is a slower process than Gzip compression, so gzip may be a better option for the compression of non-cacheable content.

Quoting Google Opensource another reliable source, I think:

While Zopfli is Deflate-compatible, Brotli is a whole new data format. This new format allows us to get 20–26% higher compression ratios over Zopfli. In our study ‘Comparison of Brotli, Deflate, Zopfli, LZMA, LZHAM and Bzip2 Compression Algorithms’ we show that Brotli is roughly as fast as zlib’s Deflate implementation. At the same time, it compresses slightly more densely than LZMA and bzip2 on the Canterbury corpus.

Of course, to get the whole context of these quotes, you should read the articles by yourself. It seems that allowing
Brotli compression is a very good choice. Let me say that this doesn't imply, at all, disallowing support for Gzip.
Which is, by the way, the most used.

Also you should check this article, .NET blog, that it is dedicated to explain the advantage and disadvantage of Brotli.

General desciption of how it works, the middleware of course.

We need to:

1. Be able to compress content with the Brotli algorithm. A magic library would do that for us.
2. Identify the scope in which the middleware is been called. More about this later.
3. Identify if the client, accept Brotli content as a response. Otherwise doesn't make any sense to send it a content that it cannot understand. By client, I mean, on this case, by any HTTP client tha can make a request to the server.
4. Learn how to send the content compressed.

So let's go for parts:

How to compress data with this algorithm.

Here we're going to make use of the official biding for python, which can be installed with:

$ pip install brotli

The official library is on google/brotli, so you can start hacking around with it if you want to.

Well, the interface to compress content with the library is pretty simple. We have

brotli.compress(string, mode=0, quality=11, lgwin=22, lgblock=0)

Compress a byte string.

Args:
  string (bytes): The input data.
  mode (int, optional): The compression mode can be MODE_GENERIC (default),
    MODE_TEXT (for UTF-8 format text input) or MODE_FONT (for WOFF 2.0).
  quality (int, optional): Controls the compression-speed vs compression-
    density tradeoff. The higher the quality, the slower the compression.
    Range is 0 to 11. Defaults to 11.
  lgwin (int, optional): Base 2 logarithm of the sliding window size. Range
    is 10 to 24. Defaults to 22.
  lgblock (int, optional): Base 2 logarithm of the maximum input block size.
    Range is 16 to 24. If set to 0, the value will be set based on the
    quality. Defaults to 0.

We are going to be using the defaults option for simplicity. Only need a byte like object to be passed as argument, and tada 😊.

Identifying the scope.

As we saw on the previous post, from the ASGI specs we know that

It takes a scope, which is a dict containing details about the specific connection.

In our case we need to know if the request been passed is an HTTP request, and only with this value we're going to work.
For that, the scope dictionary have a key called type, which specify the protocol that is incoming.

Identify if the client accept Brotli content

Looking at Accept-Encoding (MDN), we know that we can check this by looking at the value of the header Accept-Encoding. Where br correspond to the value for Brotli compression, so if the client send us a header Accept-Encoding: br we can send it the response encoded with the Brotli compression. Pretty simple, right? Also let me tell you that this header, most of the time, doesn't come with only one value, more likely it comes like this Accept-Encoding: br, gzip, deflate. Specifying more than one compression accepted by the client.

Time to code

The code that you're going to see below takes a lot of ideas from the code on FastAPI, related to GZipMiddleware. Any way, it helps me a lot to learn of how to implement a simple middleware, that is pretty usefull. Learning through imitation is how babies learn, and right now, if you're new to this content, you are pretty much like a baby, learning what is going on with all this.

Here we go, let's split the code in part so you can understand first by parts, and later you'll be able to solve the puzzle by yourself. I cannot put anything on your hand, so I will left some part of the code without explanation on purpose. In this way you should do the extra work, studying by yourself.

from starlette.types import ASGIApp, Scope, Receive, Send, Message
from starlette.datastructures import Headers, MutableHeaders
from fastapi import FastAPI

app = FastAPI()

app.add_middleware(BrotliMiddleware, minimum_size=100)

@app.get("/")
async def main():
    return {"info": "Lola"*10000}

There's nothing new here, only that we haven't define our BrotliMiddleware class.

From our last post, we know the structure of a middleware on FastAPI is not other than an ASGI middleware, so we could manage to structure it this way.

class BrotliMiddleware:
    def __init__(self, app: ASGIApp, minimum_size=100):
        self.app = app
        self.minimum_size = minimum_size

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # checking the type of the request
        if scope["type"] == "http":
            # check if the Accept-Encoding sent by the client
            headers = Headers(scope=scope)
            if "br" in headers.get("Accept-Encoding", ""):
                responder = BrotliResponser(self.app, self.minimum_size) # constructing the responder
                await responder(scope, receive, send)
                return

        await self.app(scope, receive, send)

The important part here is, how did I determine if the client accept the Brotli encoding?

headers = Headers(scope=scope)
if "br" in headers.get("Accept-Encoding", ""):
    # ... rest of the code

Well I need to be able to process the headers comming on the request, that's the first step. Remember, that this headers
are going to arrive as [(byte string name, byte string values)], so it would be very nice if we have something to handle this for us. Well, starlette provide us with a class called Headers, that will allow us to manipulate this headers in a very easy way. Again a lot of work is been simplefied thanks to starlette.

After we check that the client accept our encoding, we need to make all the hard part, which in our case means, to encode the content. If you take a look at the snippets above, you can notice that the encoding will be handled by a class called BrotliResponser, which by the way, is another ASGI class. Now let's see the real magic:

class BrotliResponser:
    def __init__(self, app: ASGIApp, minimum_size: int):
        self.app = app
        self.minimum_size = minimum_size
        self.send = None
        self.started = False
        self.initial_message: Message = {}

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        self.send = send
        await self.app(scope, receive, self.send_with_brotli)

    async def send_with_brotli(self, message: Message):
        message_type = message["type"]
        if message_type == "http.response.start":
            self.initial_message = message
        elif message_type == "http.response.body" and not self.started:
            self.started = True
            body = message.get("body", b"")
            more_body = message.get("more_body", False)
            if len(body) < self.minimum_size and not more_body:
                # Don't apply Brotli to the content sent to the response
                await self.send(self.initial_message)
                await self.send(message)
            elif not more_body:
                # Go go Brotli
                # Get body compressed
                body = brotli.compress(body)
                # Add Content-Encoding, Content-Length and Accept-Encoding
                # Why a mutable header?
                headers = MutableHeaders(raw=self.initial_message["headers"])
                headers["Content-Encoding"] = "br"
                headers.add_vary_header("Accept-Encoding")
                headers["Content-Length"] = str(len(body))

                # Body
                message["body"] = body

                await self.send(self.initial_message)
                await self.send(message)
            else:
                # streaming response, I think
                # how it works the streaming response with Brotli
                headers = MutableHeaders(raw=self.initial_message["headers"])
                headers["Content-Encoding"] = "br"
                headers.add_vary_header("Accept-Encoding")
                del headers["Content-Length"]

                # writing body
                body = brotli.compress(body)
                message["body"] = body

                await self.send(self.initial_message)
                await self.send(message)

        elif message_type == "http.response.body":
            # Remaining streaming Brotli response
            body = message.get("body", b"")
            more_body = message.get("more_body", False)

            message["body"] = brotli.compress(body)

            await self.send(message)

Ufff!!! This is too much right? How can I split this?, so you could understand the mess. I'll try my best. Let's go for the easy part at first, the initialization:

class BrotliResponser:
    def __init__(self, app: ASGIApp, minimum_size: int):
        self.app = app
        self.minimum_size = minimum_size
        self.send = None
        self.started = False
        self.initial_message: Message = {}

Nothing weird here, only one parameter called minimum_size. Which, as you may imagine, is the minimum_size required to compress a response. Otherwise all the response will be compress, even the short ones, and that's not the behavior that we want on our application. A point to notice here, this number is completly arbitrary, so you could investigate for a better option and tell me if you find one.

Now the method that have all the tricky stuff is called send_with_brotli. This method, take as argument a dictionary which will contain the information passed back to the client, on our case the content of the response. So it will define the way we are going to send the response to the client, let's break it and see what it's inside.

We need need to identify several cases:

1. We need to store the initial message in a variable, in this way we could send the whole information to the client, and not only the body. This initial message contains information related to the whole message, so is important to no loose it. Take a look at it

{'type': 'http.response.start', 'status': 200, 'headers': [(b'content-length', b'40011'), (b'content-type', b'application/json')]}

1. When the type of message is not "http.response.body", we have two cases:
   • We haven't send any data, that translate in the variable self.started is False.
   • We already started to send the data, self.started = True.

The first case is the easiest one

        if message_type == "http.response.start":
            self.initial_message = message

The second case, part one

• In case the length of the response body is too small and ther's no more body on the response, we send the message without compression.

            if len(body) < self.minimum_size and not more_body:
                # Don't apply Brotli to the content sent to the response
                # response size too small
                await self.send(self.initial_message)
                await self.send(message)

• In case there's no more body, but the current body for the response is too large, we apply compression.

            elif not more_body:
                # Go go Brotli
                # Get body compressed
                body = brotli.compress(body)
                # Add Content-Encoding, Content-Length and Accept-Encoding
                # Why a mutable header?
                headers = MutableHeaders(raw=self.initial_message["headers"])
                headers["Content-Encoding"] = "br"
                headers.add_vary_header("Accept-Encoding")
                headers["Content-Length"] = str(len(body))

                # Body
                message["body"] = body

                await self.send(self.initial_message)
                await self.send(message)

• And for last, if the current body is large and there's more body comming in

            else:
                # streaming response
                headers = MutableHeaders(raw=self.initial_message["headers"])
                headers["Content-Encoding"] = "br"
                headers.add_vary_header("Accept-Encoding")
                del headers["Content-Length"]

                # writing body
                body = brotli.compress(body)
                message["body"] = body

                await self.send(self.initial_message)
                await self.send(message)

The second case, part two

        elif message_type == "http.response.body":
            # Remaining streaming Brotli response
            body = message.get("body", b"")
            more_body = message.get("more_body", False)

            message["body"] = brotli.compress(body)

            await self.send(message)

This last part you need to figure out by yourself.

Notes about this article and the code

This article was more like an extension of the first article on the series, Middleware on FastAPI. I think the best part of all this is that I've learn all this by simple reading the documentation of FastAPI, and browsing on the web any aditional information. The most important thing to point here is that learning through imitation is one of the easiest way to learn, I take almost anything from the code related to GZipMiddleware, but of course I need to make some adaptations. Otherwise it would be the same middleware.
So don't freak out if you don't understand a code at first sight, before quiting, try to chunk the parts so you could understand the whole thing by parts. A good lecture that explain this point really well, is one given by Raimond Hettinger, is called The mental game of Python, you should check it out. The code is with Python but the technique can be apply to any problem, even outside programming.

👋 👋 👋 👋