Add server capability to check for Brotli compressed static files

.gitattributes

@@ -1,3 +1,3 @@
 tests/data.unknown_mime_type binary
-tests/hello.txt.gz binary
+tests/hello.txt.* binary
 tests/sample.* binary

CHANGES/8062.feature

@@ -0,0 +1 @@
+Added server capability to check for static files with Brotli compression via a ``.br`` extension -- by :user:`steverep`

CONTRIBUTORS.txt

@@ -306,6 +306,7 @@ Stepan Pletnev
 Stephan Jaensch
 Stephen Cirelli
 Stephen Granade
+Steve Repsher
 Steven Seguin
 Sunghyun Hwang
 Sunit Deshpande

aiohttp/web_fileresponse.py

@@ -24,7 +24,7 @@
     HTTPPreconditionFailed,
     HTTPRequestRangeNotSatisfiable,
 )
-from .web_response import StreamResponse
+from .web_response import ContentCoding, StreamResponse
 
 __all__ = ("FileResponse",)
 
@@ -37,6 +37,14 @@
 
 NOSENDFILE: Final[bool] = bool(os.environ.get("AIOHTTP_NOSENDFILE"))
 
+# File extensions for IANA encodings that will be checked in the order defined.
+# TODO(py311): Use StrEnum conversion of ContentCoding for key type and remove .value
+ENCODING_EXTENSION: Final[dict[str, str]] = {
+    # Literal for Brotli is used until compression is supported.
+    "br": ".br",
+    ContentCoding.gzip.value: ".gz",
+}
+
 
 class FileResponse(StreamResponse):
     """A response object can be used to send files."""
@@ -122,29 +130,31 @@ async def _precondition_failed(
         return await super().prepare(request)
 
     def _get_file_path_stat_and_gzip(
-        self, check_for_gzipped_file: bool
-    ) -> Tuple[pathlib.Path, os.stat_result, bool]:
-        """Return the file path, stat result, and gzip status.
+        self, accept_encoding: str
+    ) -> Tuple[pathlib.Path, os.stat_result, Optional[str]]:
+        """Return the file path, stat result, and possible compression type.
 
         This method should be called from a thread executor
         since it calls os.stat which may block.
         """
         filepath = self._path
-        if check_for_gzipped_file:
-            gzip_path = filepath.with_name(filepath.name + ".gz")
-            try:
-                return gzip_path, gzip_path.stat(), True
-            except OSError:
-                # Fall through and try the non-gzipped file
-                pass
-
-        return filepath, filepath.stat(), False
+        for encoding, extension in ENCODING_EXTENSION.items():
+            if encoding in accept_encoding:
+                compressed_path = filepath.with_name(filepath.name + extension)
+                try:
+                    return compressed_path, compressed_path.stat(), encoding
+                except OSError:
+                    # Try the next extension
+                    pass
+
+        # Fallback to the uncompressed file
+        return filepath, filepath.stat(), None
 
     async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:
         loop = asyncio.get_event_loop()
-        check_for_gzipped_file = "gzip" in request.headers.get(hdrs.ACCEPT_ENCODING, "")
-        filepath, st, gzip = await loop.run_in_executor(
-            None, self._get_file_path_stat_and_gzip, check_for_gzipped_file
+        accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
+        filepath, st, file_compression = await loop.run_in_executor(
+            None, self._get_file_path_stat_and_gzip, accept_encoding
         )
 
         etag_value = f"{st.st_mtime_ns:x}-{st.st_size:x}"
@@ -181,7 +191,7 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
             if not ct:
                 ct = "application/octet-stream"
         else:
-            encoding = "gzip" if gzip else None
+            encoding = file_compression
 
         status = self._status
         file_size = st.st_size
@@ -261,7 +271,7 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
             self.content_type = ct
         if encoding:
             self.headers[hdrs.CONTENT_ENCODING] = encoding
-        if gzip:
+        if file_compression:
             self.headers[hdrs.VARY] = hdrs.ACCEPT_ENCODING
             # Disable compression if we are already sending
             # a compressed file since we don't want to double

aiohttp/web_response.py

@@ -53,6 +53,7 @@
     BaseClass = collections.abc.MutableMapping
 
 
+# TODO(py311): Convert to StrEnum for wider use
 class ContentCoding(enum.Enum):
     # The content codings that we have support for.
     #

docs/web_reference.rst

@@ -1755,8 +1755,9 @@ Application and Router
       system call even if the platform supports it. This can be accomplished by
       by setting environment variable ``AIOHTTP_NOSENDFILE=1``.
 
-      If a gzip version of the static content exists at file path + ``.gz``, it
-      will be used for the response.
+      If a Brotli or gzip compressed version of the static content exists at
+      the requested path with the ``.br`` or ``.gz`` extension, it will be used
+      for the response. Brotli will be preferred over gzip if both files exist.
 
       .. warning::
 

tests/test_web_sendfile_functional.py

@@ -3,7 +3,7 @@
 import pathlib
 import socket
 import zlib
-from typing import Any, Iterable
+from typing import Any, Iterable, Optional
 
 import pytest
 
@@ -224,9 +224,14 @@ async def handler(request):
     await client.close()
 
 
+@pytest.mark.parametrize(
+    ("accept_encoding", "expect_encoding"),
+    [("gzip, deflate", "gzip"), ("gzip, deflate, br", "br")],
+)
 async def test_static_file_custom_content_type_compress(
-    aiohttp_client: Any, sender: Any
+    aiohttp_client: Any, sender: Any, accept_encoding: str, expect_encoding: str
 ):
+    """Test static compressed files are returned with expected content type and encoding"""
     filepath = pathlib.Path(__file__).parent / "hello.txt"
 
     async def handler(request):
@@ -238,47 +243,60 @@ async def handler(request):
     app.router.add_get("/", handler)
     client = await aiohttp_client(app)
 
-    resp = await client.get("/")
+    resp = await client.get("/", headers={"Accept-Encoding": accept_encoding})
     assert resp.status == 200
     body = await resp.read()
     assert b"hello aiohttp\n" == body
     assert resp.headers["Content-Type"] == "application/pdf"
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert resp.headers.get("Content-Encoding") == expect_encoding
     resp.close()
     await resp.release()
     await client.close()
 
 
+@pytest.mark.parametrize(
+    ("accept_encoding", "expect_encoding"),
+    [("gzip, deflate", "gzip"), ("gzip, deflate, br", "br")],
+)
+@pytest.mark.parametrize("forced_compression", [None, web.ContentCoding.gzip])
 async def test_static_file_with_gziped_counter_part_enable_compression(
-    aiohttp_client: Any, sender: Any
+    aiohttp_client: Any,
+    sender: Any,
+    accept_encoding: str,
+    expect_encoding: str,
+    forced_compression: Optional[web.ContentCoding],
 ):
-    """Test that enable_compression does not double compress when a .gz file is also present."""
+    """Test that enable_compression does not double compress when a static compressed file is also present."""
     filepath = pathlib.Path(__file__).parent / "hello.txt"
 
     async def handler(request):
         resp = sender(filepath)
-        resp.enable_compression()
+        resp.enable_compression(forced_compression)
         return resp
 
     app = web.Application()
     app.router.add_get("/", handler)
     client = await aiohttp_client(app)
 
-    resp = await client.get("/")
+    resp = await client.get("/", headers={"Accept-Encoding": accept_encoding})
     assert resp.status == 200
     body = await resp.read()
     assert body == b"hello aiohttp\n"
     assert resp.headers["Content-Type"] == "text/plain"
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert resp.headers.get("Content-Encoding") == expect_encoding
     resp.close()
     await resp.release()
     await client.close()
 
 
+@pytest.mark.parametrize(
+    ("extension", "expect_encoding"), [(".gz", "gzip"), (".br", "br")]
+)
 async def test_static_file_with_content_encoding(
-    aiohttp_client: Any, sender: Any
+    aiohttp_client: Any, sender: Any, extension: str, expect_encoding: str
 ) -> None:
-    filepath = pathlib.Path(__file__).parent / "hello.txt.gz"
+    """Test requesting of static compressed files returns correct content type and encoding"""
+    filepath = pathlib.Path(__file__).parent / ("hello.txt" + extension)
 
     async def handler(request):
         return sender(filepath)
@@ -291,10 +309,8 @@ async def handler(request):
     assert 200 == resp.status
     body = await resp.read()
     assert b"hello aiohttp\n" == body
-    ct = resp.headers["CONTENT-TYPE"]
-    assert "text/plain" == ct
-    encoding = resp.headers["CONTENT-ENCODING"]
-    assert "gzip" == encoding
+    assert resp.headers["CONTENT-TYPE"] == "text/plain"
+    assert resp.headers["CONTENT-ENCODING"] == expect_encoding
     resp.close()
 
     await resp.release()