Add support for deflate, brotli, and zstd compression formats

docs/configuration.md

@@ -175,6 +175,59 @@ Maximum items to show in arrays/objects before truncation.
 
 Maximum string length before truncation.
 
+## Compression Support
+
+The debug toolbar middleware automatically handles compressed HTTP responses. The following compression formats are supported:
+
+### Supported Formats
+
+| Format | Encoding Header | Library | Availability |
+|--------|----------------|---------|--------------|
+| **gzip** | `gzip` | `gzip` (stdlib) | Always available |
+| **deflate** | `deflate` | `zlib` (stdlib) | Always available |
+| **Brotli** | `br` | `brotli` | Optional (install with `pip install brotli`) |
+| **Zstandard** | `zstd` | `zstandard` | Optional (install with `pip install zstandard`) |
+
+### How It Works
+
+When the middleware encounters a compressed response:
+
+1. It detects the compression format from the `Content-Encoding` header
+2. Decompresses the response body
+3. Injects the toolbar HTML
+4. Returns the modified response **uncompressed** with no `Content-Encoding` header
+
+This ensures the toolbar is correctly injected regardless of whether your application uses compression.
+
+### Multiple Encodings
+
+The middleware correctly handles comma-separated encoding values (e.g., `Content-Encoding: gzip, identity`). Encodings are processed in reverse order, as per HTTP specification (last applied encoding is first to be removed).
+
+### Optional Dependencies
+
+To enable support for Brotli and Zstandard:
+
+```bash
+# Brotli support
+pip install debug-toolbar[litestar] brotli
+
+# Zstandard support  
+pip install debug-toolbar[litestar] zstandard
+
+# Both
+pip install debug-toolbar[litestar] brotli zstandard
+```
+
+If these libraries are not installed, the middleware will gracefully skip decompression for those formats and log a debug message.
+
+### Error Handling
+
+The middleware includes robust error handling:
+
+- **Invalid compressed data**: If data claims to be compressed but isn't valid, it falls back to treating it as uncompressed
+- **UTF-8 decode errors**: If decompressed data can't be decoded as UTF-8, the response is returned as-is
+- **Missing libraries**: If optional compression libraries aren't installed, those formats are skipped gracefully
+
 ## Environment-Based Configuration
 
 Common pattern for different environments:

src/debug_toolbar/litestar/middleware.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import gzip
 import logging
 import re
 import time
+import zlib
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Literal, cast
 from uuid import uuid4
@@ -15,6 +17,17 @@
 from debug_toolbar.litestar.panels.events import collect_events_metadata
 from litestar.middleware import AbstractMiddleware
 
+# Optional compression libraries
+try:
+    import brotli
+except ImportError:
+    brotli = None  # type: ignore[assignment]
+
+try:
+    import zstandard
+except ImportError:
+    zstandard = None  # type: ignore[assignment]
+
 if TYPE_CHECKING:
     from litestar.types import (
         ASGIApp,
@@ -190,20 +203,27 @@ async def _handle_response_body(
     async def _send_html_response(self, send: Send, context: RequestContext, state: ResponseState) -> None:
         """Process and send buffered HTML response with toolbar injection."""
         full_body = b"".join(state.body_chunks)
+        content_encoding = state.headers.get("content-encoding", "")
 
         try:
             await self.toolbar.process_response(context)
-            modified_body = self._inject_toolbar(full_body, context)
+            modified_body, new_encoding = self._inject_toolbar(full_body, context, content_encoding)
             server_timing = self.toolbar.get_server_timing_header(context)
         except Exception:
             logger.debug("Toolbar processing failed, sending original response", exc_info=True)
             modified_body = full_body
+            new_encoding = content_encoding
             server_timing = None
 
+        # Build headers, excluding content-length (recalculated) and content-encoding (may have changed)
+        excluded_headers = {"content-length", "content-encoding"}
         new_headers: list[tuple[bytes, bytes]] = [
-            (k.encode(), v.encode()) for k, v in state.headers.items() if k.lower() != "content-length"
+            (k.encode(), v.encode()) for k, v in state.headers.items() if k.lower() not in excluded_headers
         ]
         new_headers.append((b"content-length", str(len(modified_body)).encode()))
+        # Only add content-encoding if we still have one (not stripped due to decompression)
+        if new_encoding:
+            new_headers.append((b"content-encoding", new_encoding.encode()))
         if server_timing:
             new_headers.append((b"server-timing", server_timing.encode()))
 
@@ -471,20 +491,86 @@ def _populate_routes_metadata(self, request: Request, context: RequestContext) -
             context.metadata["routes"] = []
             context.metadata["matched_route"] = ""
 
-    def _inject_toolbar(self, body: bytes, context: RequestContext) -> bytes:
+    def _decompress_body(self, body: bytes, encodings: list[str]) -> tuple[bytes, bool]:  # noqa: PLR0912
+        """Decompress response body based on content-encoding.
+
+        Args:
+            body: The compressed response body.
+            encodings: List of encoding formats (e.g., ["gzip", "deflate"]).
+
+        Returns:
+            Tuple of (decompressed body, success flag).
+            Success flag is True if decompression succeeded, False otherwise.
+        """
+        decompressed = False
+        # Process encodings in reverse order (last applied encoding is first to remove)
+        for encoding in reversed(encodings):
+            if encoding == "gzip":
+                try:
+                    body = gzip.decompress(body)
+                    decompressed = True
+                except (gzip.BadGzipFile, OSError):
+                    # Not valid gzip, try next encoding or decode as-is
+                    break
+            elif encoding == "deflate":
+                try:
+                    body = zlib.decompress(body)
+                    decompressed = True
+                except zlib.error:
+                    # Not valid deflate, try next encoding or decode as-is
+                    break
+            elif encoding == "br":
+                if brotli is not None:
+                    try:
+                        body = brotli.decompress(body)
+                        decompressed = True
+                    except brotli.error:  # type: ignore[attr-defined]
+                        # Not valid brotli, try next encoding or decode as-is
+                        break
+                else:
+                    # Brotli not available, can't decompress
+                    logger.debug("Brotli compression detected but brotli library not installed")
+                    break
+            elif encoding == "zstd":
+                if zstandard is not None:
+                    try:
+                        dctx = zstandard.ZstdDecompressor()
+                        body = dctx.decompress(body)
+                        decompressed = True
+                    except zstandard.ZstdError:
+                        # Not valid zstd, try next encoding or decode as-is
+                        break
+                else:
+                    # Zstandard not available, can't decompress
+                    logger.debug("Zstandard compression detected but zstandard library not installed")
+                    break
+        return body, decompressed
+
+    def _inject_toolbar(self, body: bytes, context: RequestContext, content_encoding: str = "") -> tuple[bytes, str]:
         """Inject the toolbar HTML into the response body.
 
         Args:
-            body: The original response body.
+            body: The original response body (may be compressed).
             context: The request context with collected data.
+            content_encoding: The content-encoding header value (e.g., "gzip", "deflate", "br", "zstd").
 
         Returns:
-            The modified response body with toolbar injected.
+            Tuple of (modified body, content_encoding to use).
+            If compression was decompressed, returns uncompressed body with empty encoding.
         """
+        # Handle compressed responses
+        encodings = [e.strip() for e in content_encoding.lower().split(",")] if content_encoding else []
+        body, decompressed = self._decompress_body(body, encodings)
+
         try:
             html = body.decode("utf-8")
         except UnicodeDecodeError:
-            return body
+            # Can't decode. If we successfully decompressed, return the
+            # decompressed body with no content-encoding. Otherwise, return
+            # the body as-is with the original encoding.
+            if decompressed:
+                return body, ""
+            return body, content_encoding
 
         toolbar_data = self.toolbar.get_toolbar_data(context)
         toolbar_html = self._render_toolbar(toolbar_data)
@@ -496,7 +582,10 @@ def _inject_toolbar(self, body: bytes, context: RequestContext) -> bytes:
             pattern = re.compile(re.escape(insert_before), re.IGNORECASE)
             html = pattern.sub(toolbar_html + insert_before, html, count=1)
 
-        return html.encode("utf-8")
+        # Return body as uncompressed UTF-8 with empty content-encoding.
+        # This applies to all successful toolbar injections, regardless of whether
+        # the input was originally compressed (we decompress before processing).
+        return html.encode("utf-8"), ""
 
     def _render_toolbar(self, data: dict[str, Any]) -> str:
         """Render the toolbar HTML.