Match WhiteNoise upstream compress.py (#62)

Co-authored-by: Adam Chainz <[email protected]>

CHANGELOG.md

@@ -15,6 +15,7 @@ Using the following categories, list your changes in this order:
 
 -   Drop Django 3.2 and 4.1 support.
 -   Any errors from threads in the `servestatic.compress` command are now raised.
+-   Compression code has been refactored to match upstream (WhiteNoise).
 
 ## [2.1.1] - 2024-10-27
 

src/servestatic/compress.py

@@ -69,25 +69,24 @@ def get_extension_re(extensions):
     def should_compress(self, filename):
         return not self.extension_re.search(filename)
 
-    def _lazy_compress(self, path):
+    def compress(self, path):
+        filenames = []
         with open(path, "rb") as f:
             stat_result = os.fstat(f.fileno())
             data = f.read()
         size = len(data)
         if self.use_brotli:
             compressed = self.compress_brotli(data)
             if self.is_compressed_effectively("Brotli", path, size, compressed):
-                yield self.write_data(path, compressed, ".br", stat_result)
+                filenames.append(self.write_data(path, compressed, ".br", stat_result))
             else:
                 # If Brotli compression wasn't effective gzip won't be either
-                return
+                return filenames
         if self.use_gzip:
             compressed = self.compress_gzip(data)
             if self.is_compressed_effectively("Gzip", path, size, compressed):
-                yield self.write_data(path, compressed, ".gz", stat_result)
-
-    def compress(self, path):
-        return list(self._lazy_compress(path))
+                filenames.append(self.write_data(path, compressed, ".gz", stat_result))
+        return filenames
 
     @staticmethod
     def compress_gzip(data):

src/servestatic/storage.py

@@ -38,11 +38,13 @@ def post_process(self, paths: dict[str, Any], dry_run: bool = False, **options:
         self.compressor = self.create_compressor(extensions=extensions, quiet=True)
 
         def _compress_path(path: str) -> Generator[tuple[str, str, bool]]:
+            compressed: list[tuple[str, str, bool]] = []
             full_path = self.path(path)
             prefix_len = len(full_path) - len(path)
             for compressed_path in self.compressor.compress(full_path):
                 compressed_name = compressed_path[prefix_len:]
-                yield (path, compressed_name, True)
+                compressed.append((path, compressed_name, True))
+            return compressed
 
         with ThreadPoolExecutor() as executor:
             futures = (executor.submit(_compress_path, path) for path in paths if self.compressor.should_compress(path))
@@ -185,25 +187,20 @@ def compress_files(self, paths):
         extensions = getattr(settings, "SERVESTATIC_SKIP_COMPRESS_EXTENSIONS", None)
         self.compressor = self.create_compressor(extensions=extensions, quiet=True)
 
-        def _compress_path(path: str) -> Generator[tuple[str, str]]:
+        def _compress_path(path: str) -> list[tuple[str, str]]:
+            compressed: list[tuple[str, str]] = []
             full_path = self.path(path)
             prefix_len = len(full_path) - len(path)
             for compressed_path in self.compressor.compress(full_path):
                 compressed_name = compressed_path[prefix_len:]
-                yield (path, compressed_name)
+                compressed.append((path, compressed_name))
+            return compressed
 
         with ThreadPoolExecutor() as executor:
             futures = (executor.submit(_compress_path, path) for path in paths if self.compressor.should_compress(path))
             for future in as_completed(futures):
                 yield from future.result()
 
-    def _compress_one(self, name: str) -> list[tuple[str, str]]:
-        compressed: list[tuple[str, str]] = []
-        path = self.path(name)
-        prefix_len = len(path) - len(name)
-        compressed.extend((name, compressed_path[prefix_len:]) for compressed_path in self.compressor.compress(path))
-        return compressed
-
     def make_helpful_exception(self, exception, name):
         """
         If a CSS file contains references to images, fonts etc that can't be found