FastapI로 Brotli 미들웨어 구축

최초로 저의 blog에 발표가 됐습니다.
지금, 내가 지난 글에서 약속한 바와 같이, 나는 더욱 복잡한 중간부품을 구축할 것이다.나는 google/brotli 실현에 흥미가 없다는 것을 감안하여 대부분의 업무는 Brotli algorithm이라는 라이브러리에서 완성될 것이다.이와 반대로, 내가 여기에 있는 주요 목적은 중간부품을 실현하는 것이다. 그 행위는 FastAPI 문서의 GZipMiddleware와 유사하다.

Brotli 알고리즘 정보
너는 내가 압축 알고리즘 전문가가 아니라고 상상할 수 있다. 그래서 내가 말하고자 하는 모든 것은 웹 페이지를 조회한 결과이다.
네트워크 개발에 대한 MDN의 가장 신뢰할 수 있는 참조 중 하나:

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.

Google Opensource의 또 다른 믿을 만한 출처를 인용하여 다음과 같이 생각합니다.

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.

물론 이 인용어의 전체 내용을 이해하려면, 너는 반드시 스스로 이 문장을 읽어야 한다.보아하니
Brotli 압축이 적합합니다.이것은 Gzip을 지원하는 것을 허락하지 않는다는 것을 의미하지 않는다.
겸사겸사 한마디 하자면, 이것은 가장 자주 사용하는 것이다.
Brotli의 장점과 단점을 전문적으로 설명하는 .NET blog을 보셔야 합니다.

그것이 어떻게 일을 하는지에 대한 일반적인 설명은 당연히 중간부품이다.
필요한 사항:

은 Brotli 알고리즘으로 컨텐츠를 압축할 수 있습니다.마법 도서관은 우리를 도와 이 점을 해낼 수 있다.

표지 호출 중간부품의 범위.이따가 다시 이야기합시다.

은 고객이 Brotli 컨텐츠를 응답으로 받아들일지 여부를 결정합니다.그렇지 않으면 이해할 수 없는 내용을 보내는 것은 무의미하다.클라이언트, 내 말은 이 경우 모든 HTTP 클라이언트가 서버에 요청을 할 수 있다는 것이다.

압축된 내용을 어떻게 보내는지 배웁니다.

그러면 부품을 살펴보겠습니다.

어떻게 이런 알고리즘으로 데이터를 압축합니까?
여기서, 우리는python의 공식 입찰을 이용하여 설치할 수 있습니다.

$ pip install brotli

공식 도서관의 전화는 google/brotli이기 때문에 원한다면 해킹을 시작할 수 있다.
응, 라이브러리로 내용을 압축하는 인터페이스는 매우 간단해.우리

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.

간단하게 보기 위해서 기본 옵션을 사용할 것입니다.바이트와 같은 대상을 매개 변수로 전달하고tada😊.

범위를 정하다.
우리가 이전 글에서 보듯이 ASGI 규범에서 우리는

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

우리의 예에서 전달된 요청이 HTTP 요청인지 알아야 합니다. 이 값을 사용해야만 작업을 할 수 있습니다.
이를 위해, 역할 영역 사전에는 type이라는 키가 있는데, 전송된 프로토콜을 지정합니다.

고객이 Brotli 컨텐츠를 수락했는지 확인
Accept-Encoding (MDN)을 보십시오. 우리는 머리가 인코딩된 값을 보면서 이 점을 검사할 수 있다는 것을 알고 있습니다.그 중에서 Br는 Brotli 압축 값에 대응하기 때문에 클라이언트가 우리에게 머리 Accept-Encoding: br을 보내면 Brotli 압축 코드로 응답을 보낼 수 있다.아주 간단하죠?이 제목은 대다수 상황에서 단지 하나의 값만 있는 것이 아니라 Accept-Encoding: br, gzip, deflate일 수도 있다는 것을 알려드리겠습니다.클라이언트가 받아들일 여러 개의 압축을 지정합니다.

인코딩할 때가 됐어요.
다음은 GZipMiddleware와 관련된 FastapI의 코드에서 많은 아이디어를 흡수한 코드입니다.어쨌든, 어떻게 매우 유용한 간단한 중간부품을 실현하는지 배우는 것은 나에게 매우 큰 도움이 된다.모방을 통해 배우는 것은 아기가 배우는 방식이다. 그러나 지금 이 내용에 대해 익숙하지 않다면, 당신은 마치 아기처럼 이 모든 것을 배우고 있다.
이제 코드를 몇 부분으로 나누자. 그러면 너는 먼저 부분적으로 이해할 수 있고, 그런 후에 너는 스스로 이 난제를 해결할 수 있다.나는 어떤 것도 너의 손에 놓을 수 없기 때문에, 나는 일부러 해석하지 않은 코드를 남길 것이다.이렇게 하면 너는 마땅히 별도의 일을 하고 스스로 공부해야 한다.

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}

여기에는 새로운 것이 없습니다. 다만 Brotlimidware 종류는 아직 정의하지 않았습니다.
지난 글에서Fast API의 중간부품의 구조가 ASGI 중간부품이 아니라는 것을 알았기 때문에 우리는 이런 방식으로 그것을 구성할 수 있다.

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)

여기서 중요한 부분은 클라이언트가 Brotli 인코딩을 받아들일지 여부를 어떻게 확인합니까?

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

요청한 헤더 파일을 처리할 수 있는 것이 첫 번째 단계입니다.이것 괜찮아요?
우리는 [(byte string name, byte string values)]의 신분으로 도착할 것이기 때문에 만약 우리가 이 일을 처리할 수 있다면 매우 좋을 것이다.starlette은 Headers라는 클래스를 제공합니다. 이것은 우리가 매우 간단한 방식으로 이 Headers를 조작할 수 있도록 합니다.starlette으로 인해 많은 작업이 다시 간소화되었다.
고객이 우리의 인코딩을 받아들였는지 확인한 후에 우리는 모든 어려운 부분을 해야 한다. 우리의 예에서 이것은 내용을 인코딩하는 것을 의미한다.위의 코드 세션을 보면 인코딩은 BrotliResponser이라는 클래스로 처리됩니다. 참고로 다른 ASGI 클래스입니다.이제 진정한 마법을 보자.

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!!!이건 너무하잖아, 그렇지?내가 어떻게 그걸 분리할 수 있겠어?이렇게 하면 너는 이 엉망진창을 이해할 수 있다.최선을 다하겠습니다.간단한 섹션부터 살펴보고 초기화:

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 = {}

여기에는 이상한 것이 없다. 단지 minimum_size이라는 매개 변수만 있다.이것은 압축 응답에 필요한 minimum_size이라고 상상할 수 있다.그렇지 않으면, 모든 응답이 아주 짧은 응답이라도 압축됩니다. 이것은 우리가 원하는 프로그램의 행동이 아닙니다.여기서 주의해야 할 점은 이 숫자는 완전히 임의이기 때문에 더 좋은 선택을 조사할 수 있다. 만약 네가 찾았다면 나에게 말해라.
현재 모든 까다로운 내용을 포함하는 방법은 send_with_brotli이라고 불린다.이 방법은 사전을 매개 변수로 하여 클라이언트에 되돌아오는 정보를 포함하고 우리의 예에서 응답하는 내용을 포함한다.그것은 우리가 응답을 클라이언트에게 보내는 방식을 정의하고, 그 안에 무엇이 있는지 깨뜨려 보자.
우리는 몇 가지 사례를 확정해야 한다.

우리는 초기 메시지를 하나의 변수에 저장해야 한다. 그러면 우리는 전체 정보를 클라이언트에게 보낼 수 있고 주체에게만 보낼 수 있다.이 초기 메시지는 전체 메시지와 관련된 정보를 포함하기 때문에 잃어버리지 않는 것이 매우 중요하다.보다

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

메시지 유형이 "http.response.body"이 아닌 경우

우리는 아직 어떤 데이터도 보내지 않았고 변수 self.started에서 False으로 전환했다.

데이터 전송을 시작했습니다. self.started = True.

첫 번째 상황이 제일 쉬워요.

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

두 번째 사례, 첫 번째 부분

응답 주체의 길이가 너무 작고 응답에 더 많은 주체가 없으면 우리는 압축하지 않은 상황에서 메시지를 보낼 것이다.

            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)

더 많은 주체가 없지만 현재의 주체 응답이 너무 크면 우리는 압축을 응용할 것이다.

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

마지막 부분은 네가 스스로 해결해야 한다.

본문과 코드에 대한 주석
이 문장은 시리즈의 첫 번째 문장의 연장선상에 가깝다.나는 이 모든 가장 좋은 부분은 내가 FastAPI의 문서를 간단하게 읽고 인터넷에서 어떤 전통적인 정보를 조회해서 이 모든 것을 이해하는 것이라고 생각한다.여기서 지적하고자 하는 가장 중요한 점은 모방 학습을 통해 가장 간단한 학습 방식 중의 하나이다. 나는 GZip Middleware와 관련된 코드에서 거의 모든 내용을 얻었지만 물론 수정이 필요하다.그렇지 않으면 그것은 같은 중간부품이 될 것이다.
그래서 만약에 코드를 한눈에 모르면 당황하지 말고 종료하기 전에 코드를 몇 부분으로 나누어 보세요. 그러면 코드를 한 부분 한 부분 이해할 수 있습니다.Raimond Hettinger가 좋은 강연을 했습니다. 이 강연은 이 점을 잘 설명했습니다. The mental game of Python이라고 합니다. 가보셔야 합니다.코드는 파이톤으로 작성되었지만, 이런 기술은 어떠한 문제, 심지어 외부 프로그래밍에도 적용될 수 있다.
👋 👋 👋 👋