swarmauri_middleware_gzipcompression

Middleware for adding gzip compression to FastAPI responses.

Purpose

This package provides a middleware that automatically compresses outgoing responses using gzip encoding. It ensures that responses are only compressed when supported by the client and when the content type is appropriate for compression.

Installation

pip

pip install swarmauri_middleware_gzipcompression

Poetry

poetry add swarmauri_middleware_gzipcompression

uv

uv pip install swarmauri_middleware_gzipcompression

How it works

The middleware mirrors FastAPI's BaseHTTPMiddleware contract and can be plugged into an application using the standard @app.middleware("http") hook. Compression is only applied when all of the following are true:

• The downstream component returns a fastapi.Response (non-Response results pass through unchanged).
• The response has a compressible media type such as application/json or any text/* content type.
• The client opts into compression by sending an Accept-Encoding header that includes gzip.
• The response is not already marked as gzip encoded.

Example

The snippet below simulates FastAPI's middleware pipeline and demonstrates how gzip compression is applied when the client opts in to receive gzip-encoded responses.

import asyncio
import gzip
from fastapi import Request, Response

from swarmauri_middleware_gzipcompression import GzipCompressionMiddleware

middleware = GzipCompressionMiddleware()

async def _handle_request() -> None:
    scope = {
        "type": "http",
        "method": "GET",
        "headers": [(b"accept-encoding", b"gzip")],
    }

    async def receive() -> dict:
        return {"type": "http.request", "body": b""}

    request = Request(scope, receive=receive)

    async def call_next(_: Request) -> Response:
        return Response(
            content='{"message":"Hello, gzip!"}',
            media_type="application/json",
        )

    response = await middleware.dispatch(request, call_next)

    assert response.headers.get("Content-Encoding") == "gzip"
    body = gzip.decompress(response.body).decode("utf-8")
    assert body == '{"message":"Hello, gzip!"}'

def run_example() -> None:
    asyncio.run(_handle_request())

if __name__ == "__main__":
    run_example()

Want to help?

If you want to contribute to swarmauri-sdk, read up on our guidelines for contributing that will help you get started.