48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
from starlette.datastructures import Headers
|
|
from starlette.middleware.gzip import GZipResponder, IdentityResponder
|
|
from starlette.types import ASGIApp, Receive, Scope, Send
|
|
|
|
|
|
def _quality(accept_encoding: str, coding: str) -> float:
|
|
qualities: dict[str, float] = {}
|
|
for entry in accept_encoding.split(","):
|
|
parts = [part.strip() for part in entry.split(";")]
|
|
name = parts[0].lower()
|
|
if not name:
|
|
continue
|
|
quality = 1.0
|
|
for parameter in parts[1:]:
|
|
key, separator, value = parameter.partition("=")
|
|
if separator and key.strip().lower() == "q":
|
|
try:
|
|
quality = float(value.strip())
|
|
except ValueError:
|
|
quality = 0.0
|
|
qualities[name] = quality
|
|
return qualities.get(coding, qualities.get("*", 0.0))
|
|
|
|
|
|
class NegotiatedGZipMiddleware:
|
|
"""Compress substantial responses only when the client permits gzip."""
|
|
|
|
def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 6):
|
|
self.app = app
|
|
self.minimum_size = minimum_size
|
|
self.compresslevel = compresslevel
|
|
|
|
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
|
if scope["type"] != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
accepted = Headers(scope=scope).get("Accept-Encoding", "")
|
|
path = scope.get("path", "").rsplit("/", 1)[-1]
|
|
immutable_runtime = path.startswith("runtime-") and path.endswith(".js")
|
|
if _quality(accepted, "gzip") > 0 and not immutable_runtime:
|
|
responder: ASGIApp = GZipResponder(
|
|
self.app, self.minimum_size, compresslevel=self.compresslevel
|
|
)
|
|
else:
|
|
responder = IdentityResponder(self.app, self.minimum_size)
|
|
await responder(scope, receive, send)
|