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.rst

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

aiohttp/web_fileresponse.py

@@ -2,12 +2,15 @@
 import mimetypes
 import os
 import pathlib
+import sys
+from contextlib import suppress
 from typing import (
     IO,
     TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
+    Dict,
     Final,
     Optional,
     Tuple,
@@ -37,6 +40,14 @@
 
 NOSENDFILE: Final[bool] = bool(os.environ.get("AIOHTTP_NOSENDFILE"))
 
+if sys.version_info < (3, 9):
+    mimetypes.encodings_map[".br"] = "br"
+
+# File extension to IANA encodings map that will be checked in the order defined.
+ENCODING_EXTENSIONS: Final[Dict[str, str]] = {
+    ext: mimetypes.encodings_map[ext] for ext in (".br", ".gz")
+}
+
 
 class FileResponse(StreamResponse):
     """A response object can be used to send files."""
@@ -121,34 +132,36 @@ async def _precondition_failed(
         self.content_length = 0
         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.
+    def _get_file_path_stat_encoding(
+        self, accept_encoding: str
+    ) -> Tuple[pathlib.Path, os.stat_result, Optional[str]]:
+        """Return the file path, stat result, and encoding.
+
+        If an uncompressed file is returned, the encoding is set to
+        :py:data:`None`.
 
         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
+        file_path = self._path
+        for file_extension, file_encoding in ENCODING_EXTENSIONS.items():
+            if file_encoding not in accept_encoding:
+                continue
+
+            compressed_path = file_path.with_suffix(file_path.suffix + file_extension)
+            with suppress(OSError):
+                return compressed_path, compressed_path.stat(), file_encoding
 
-        return filepath, filepath.stat(), False
+        # Fallback to the uncompressed file
+        return file_path, file_path.stat(), None
 
     async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:
         loop = asyncio.get_event_loop()
         # Encoding comparisons should be case-insensitive
         # https://www.rfc-editor.org/rfc/rfc9110#section-8.4.1
-        check_for_gzipped_file = (
-            "gzip" in request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
-        )
-        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()
+        file_path, st, file_encoding = await loop.run_in_executor(
+            None, self._get_file_path_stat_encoding, accept_encoding
         )
 
         etag_value = f"{st.st_mtime_ns:x}-{st.st_size:x}"
@@ -181,11 +194,11 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
 
         ct = None
         if hdrs.CONTENT_TYPE not in self.headers:
-            ct, encoding = mimetypes.guess_type(str(filepath))
+            ct, encoding = mimetypes.guess_type(str(file_path))
             if not ct:
                 ct = "application/octet-stream"
         else:
-            encoding = "gzip" if gzip else None
+            encoding = file_encoding
 
         status = self._status
         file_size = st.st_size
@@ -265,7 +278,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_encoding:
             self.headers[hdrs.VARY] = hdrs.ACCEPT_ENCODING
             # Disable compression if we are already sending
             # a compressed file since we don't want to double
@@ -289,7 +302,7 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
         if count == 0 or must_be_empty_body(request.method, self.status):
             return await super().prepare(request)
 
-        fobj = await loop.run_in_executor(None, filepath.open, "rb")
+        fobj = await loop.run_in_executor(None, file_path.open, "rb")
         if start:  # be aware that start could be None or int=0 here.
             offset = start
         else:

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.py

@@ -21,7 +21,7 @@ def test_using_gzip_if_header_present_and_file_available(loop: Any) -> None:
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.with_name.return_value = gz_filepath
+    filepath.with_suffix.return_value = gz_filepath
 
     file_sender = FileResponse(filepath)
     file_sender._path = filepath
@@ -42,7 +42,7 @@ def test_gzip_if_header_not_present_and_file_available(loop: Any) -> None:
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.with_name.return_value = gz_filepath
+    filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
@@ -64,7 +64,7 @@ def test_gzip_if_header_not_present_and_file_not_available(loop: Any) -> None:
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.with_name.return_value = gz_filepath
+    filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
@@ -88,7 +88,7 @@ def test_gzip_if_header_present_and_file_not_available(loop: Any) -> None:
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.with_name.return_value = gz_filepath
+    filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 

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,58 @@ 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.get("Content-Encoding") == expect_encoding
     assert resp.headers["Content-Type"] == "application/pdf"
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert await resp.read() == b"hello aiohttp\n"
     resp.close()
     await resp.release()
     await client.close()
 
 
-async def test_static_file_with_gziped_counter_part_enable_compression(
-    aiohttp_client: Any, sender: Any
+@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_encoding_and_enable_compression(
+    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.get("Content-Encoding") == expect_encoding
     assert resp.headers["Content-Type"] == "text/plain"
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert await resp.read() == b"hello aiohttp\n"
     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 / f"hello.txt{extension}"
 
     async def handler(request):
         return sender(filepath)
@@ -289,12 +305,9 @@ async def handler(request):
 
     resp = await client.get("/")
     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 expect_encoding == resp.headers["CONTENT-ENCODING"]
+    assert "text/plain" == resp.headers["CONTENT-TYPE"]
+    assert b"hello aiohttp\n" == await resp.read()
     resp.close()
 
     await resp.release()