[3.10] Add server capability to check for Brotli compressed static files

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,6 +2,9 @@
 import mimetypes
 import os
 import pathlib
+import sys
+from contextlib import suppress
+from types import MappingProxyType
 from typing import (  # noqa
     IO,
     TYPE_CHECKING,
@@ -40,6 +43,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 = MappingProxyType(
+    {ext: mimetypes.encodings_map[ext] for ext in (".br", ".gz")}
+)
+
 
 class FileResponse(StreamResponse):
     """A response object can be used to send files."""
@@ -124,34 +135,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}"
@@ -183,12 +196,12 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
             return await self._not_modified(request, etag_value, last_modified)
 
         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"
             should_set_ct = True
         else:
-            encoding = "gzip" if gzip else None
+            encoding = file_encoding
             should_set_ct = False
 
         status = self._status
@@ -269,7 +282,7 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
             self.content_type = ct  # type: ignore[assignment]
         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
@@ -293,7 +306,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

@@ -52,6 +52,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

@@ -1846,8 +1846,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

@@ -20,7 +20,7 @@ def test_using_gzip_if_header_present_and_file_available(loop) -> 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
@@ -41,7 +41,7 @@ def test_gzip_if_header_not_present_and_file_available(loop) -> 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
 
@@ -63,7 +63,7 @@ def test_gzip_if_header_not_present_and_file_not_available(loop) -> 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
 
@@ -87,7 +87,7 @@ def test_gzip_if_header_present_and_file_not_available(loop) -> 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

@@ -10,6 +10,11 @@
 import aiohttp
 from aiohttp import web
 
+try:
+    import brotlicffi as brotli
+except ImportError:
+    import brotli
+
 try:
     import ssl
 except ImportError:
@@ -27,9 +32,14 @@ def hello_txt(request, tmp_path_factory) -> pathlib.Path:
     indirect parameter can be passed with an encoding to get a compressed path.
     """
     txt = tmp_path_factory.mktemp("hello-") / "hello.txt"
-    hello = {None: txt, "gzip": txt.with_suffix(f"{txt.suffix}.gz")}
+    hello = {
+        None: txt,
+        "gzip": txt.with_suffix(f"{txt.suffix}.gz"),
+        "br": txt.with_suffix(f"{txt.suffix}.br"),
+    }
     hello[None].write_bytes(HELLO_AIOHTTP)
     hello["gzip"].write_bytes(gzip.compress(HELLO_AIOHTTP))
+    hello["br"].write_bytes(brotli.compress(HELLO_AIOHTTP))
     encoding = getattr(request, "param", None)
     return hello[encoding]
 
@@ -220,7 +230,7 @@ async def handler(request):
     await client.close()
 
 
-@pytest.mark.parametrize("hello_txt", ["gzip"], indirect=True)
+@pytest.mark.parametrize("hello_txt", ["gzip", "br"], indirect=True)
 async def test_static_file_custom_content_type(
     hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
 ) -> None:
@@ -245,8 +255,16 @@ 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(
-    hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
+    hello_txt: pathlib.Path,
+    aiohttp_client: Any,
+    sender: Any,
+    accept_encoding: str,
+    expect_encoding: str,
 ):
     """Test that custom type with encoding is returned for unencoded requests."""
 
@@ -259,21 +277,27 @@ 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
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert resp.headers.get("Content-Encoding") == expect_encoding
     assert resp.headers["Content-Type"] == "application/pdf"
     assert await resp.read() == HELLO_AIOHTTP
     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_encoding_and_enable_compression(
     hello_txt: pathlib.Path,
     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 an encoded file is also present."""
@@ -287,9 +311,9 @@ 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
-    assert resp.headers.get("Content-Encoding") == "gzip"
+    assert resp.headers.get("Content-Encoding") == expect_encoding
     assert resp.headers["Content-Type"] == "text/plain"
     assert await resp.read() == HELLO_AIOHTTP
     resp.close()
@@ -298,7 +322,7 @@ async def handler(request):
 
 
 @pytest.mark.parametrize(
-    ("hello_txt", "expect_encoding"), [["gzip"] * 2], indirect=["hello_txt"]
+    ("hello_txt", "expect_encoding"), [["gzip"] * 2, ["br"] * 2], indirect=["hello_txt"]
 )
 async def test_static_file_with_content_encoding(
     hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any, expect_encoding: str