hatch fmt --formatter

scripts/generate_default_media_types.py

@@ -66,10 +66,7 @@ def default_types() -> dict[str, str]:
 
 def get_default_types_function() -> str:
     types_map = get_types_map()
-    lines = [
-        f'        "{suffix}": "{media_type}",'
-        for suffix, media_type in types_map.items()
-    ]
+    lines = [f'        "{suffix}": "{media_type}",' for suffix, media_type in types_map.items()]
     return FUNCTION_TEMPLATE.format(entries="\n".join(lines))
 
 

src/servestatic/asgi.py

@@ -45,8 +45,7 @@ async def __call__(self, scope, receive, send):
         # Convert ASGI headers into WSGI headers. Allows us to reuse all of our WSGI
         # header logic inside of aget_response().
         wsgi_headers = {
-            "HTTP_" + key.decode().upper().replace("-", "_"): value.decode()
-            for key, value in scope["headers"]
+            "HTTP_" + key.decode().upper().replace("-", "_"): value.decode() for key, value in scope["headers"]
         }
 
         # Get the ServeStatic file response

src/servestatic/base.py

@@ -202,9 +202,7 @@ def get_static_file(self, path, url, stat_cache=None):
 
     def add_mime_headers(self, headers, path, url):
         media_type = self.media_types.get_type(path)
-        params = (
-            {"charset": str(self.charset)} if media_type.startswith("text/") else {}
-        )
+        params = {"charset": str(self.charset)} if media_type.startswith("text/") else {}
         headers.add_header("Content-Type", str(media_type), **params)
 
     def add_cache_headers(self, headers, path, url):
@@ -238,9 +236,5 @@ def redirect(self, from_url, to_url):
         else:
             msg = f"Cannot handle redirect: {from_url} > {to_url}"
             raise ValueError(msg)
-        headers = (
-            {"Cache-Control": f"max-age={self.max_age}, public"}
-            if self.max_age is not None
-            else {}
-        )
+        headers = {"Cache-Control": f"max-age={self.max_age}, public"} if self.max_age is not None else {}
         return Redirect(relative_url, headers=headers)

src/servestatic/compress.py

@@ -52,9 +52,7 @@ class Compressor:
         "wmv",
     )
 
-    def __init__(
-        self, extensions=None, use_gzip=True, use_brotli=True, log=print, quiet=False
-    ):
+    def __init__(self, extensions=None, use_gzip=True, use_brotli=True, log=print, quiet=False):
         if extensions is None:
             extensions = self.SKIP_COMPRESS_EXTENSIONS
         self.extension_re = self.get_extension_re(extensions)
@@ -66,9 +64,7 @@ def __init__(
     def get_extension_re(extensions):
         if not extensions:
             return re.compile("^$")
-        return re.compile(
-            rf'\.({"|".join(map(re.escape, extensions))})$', re.IGNORECASE
-        )
+        return re.compile(rf'\.({"|".join(map(re.escape, extensions))})$', re.IGNORECASE)
 
     def should_compress(self, filename):
         return not self.extension_re.search(filename)
@@ -98,9 +94,7 @@ def compress_gzip(data):
         output = BytesIO()
         # Explicitly set mtime to 0 so gzip content is fully determined
         # by file content (0 = "no timestamp" according to gzip spec)
-        with gzip.GzipFile(
-            filename="", mode="wb", fileobj=output, compresslevel=9, mtime=0
-        ) as gz_file:
+        with gzip.GzipFile(filename="", mode="wb", fileobj=output, compresslevel=9, mtime=0) as gz_file:
             gz_file.write(data)
         return output.getvalue()
 
@@ -116,9 +110,7 @@ def is_compressed_effectively(self, encoding_name, path, orig_size, data):
             ratio = compressed_size / orig_size
             is_effective = ratio <= 0.95
         if is_effective:
-            self.log(
-                f"{encoding_name} compressed {path} ({orig_size // 1024}K -> {compressed_size // 1024}K)"
-            )
+            self.log(f"{encoding_name} compressed {path} ({orig_size // 1024}K -> {compressed_size // 1024}K)")
         else:
             self.log(f"Skipping {path} ({encoding_name} compression not effective)")
         return is_effective
@@ -144,9 +136,7 @@ def main(argv=None):
         "'.gz' and '.br' suffixes (as long as this results in a "
         "smaller file)"
     )
-    parser.add_argument(
-        "-q", "--quiet", help="Don't produce log output", action="store_true"
-    )
+    parser.add_argument("-q", "--quiet", help="Don't produce log output", action="store_true")
     parser.add_argument(
         "--no-gzip",
         help="Don't produce gzip '.gz' files",
@@ -164,10 +154,7 @@ def main(argv=None):
     parser.add_argument(
         "extensions",
         nargs="*",
-        help=(
-            "File extensions to exclude from compression "
-            + f"(default: {default_exclude})"
-        ),
+        help=("File extensions to exclude from compression " + f"(default: {default_exclude})"),
         default=Compressor.SKIP_COMPRESS_EXTENSIONS,
     )
     args = parser.parse_args(argv)

src/servestatic/middleware.py

@@ -47,9 +47,7 @@ class ServeStaticMiddleware(ServeStaticBase):
 
     def __init__(self, get_response=None, settings=django_settings):
         if not iscoroutinefunction(get_response):
-            msg = (
-                "ServeStaticMiddleware requires an async compatible version of Django."
-            )
+            msg = "ServeStaticMiddleware requires an async compatible version of Django."
             raise ValueError(msg)
         markcoroutinefunction(self)
 
@@ -60,9 +58,7 @@ def __init__(self, get_response=None, settings=django_settings):
         allow_all_origins = getattr(settings, "SERVESTATIC_ALLOW_ALL_ORIGINS", True)
         charset = getattr(settings, "SERVESTATIC_CHARSET", "utf-8")
         mimetypes = getattr(settings, "SERVESTATIC_MIMETYPES", None)
-        add_headers_function = getattr(
-            settings, "SERVESTATIC_ADD_HEADERS_FUNCTION", None
-        )
+        add_headers_function = getattr(settings, "SERVESTATIC_ADD_HEADERS_FUNCTION", None)
         self.index_file = getattr(settings, "SERVESTATIC_INDEX_FILE", None)
         immutable_file_test = getattr(settings, "SERVESTATIC_IMMUTABLE_FILE_TEST", None)
         self.use_finders = getattr(settings, "SERVESTATIC_USE_FINDERS", debug)
@@ -73,9 +69,7 @@ def __init__(self, get_response=None, settings=django_settings):
         )
         self.static_prefix = getattr(settings, "SERVESTATIC_STATIC_PREFIX", None)
         self.static_root = getattr(settings, "STATIC_ROOT", None)
-        self.keep_only_hashed_files = getattr(
-            django_settings, "SERVESTATIC_KEEP_ONLY_HASHED_FILES", False
-        )
+        self.keep_only_hashed_files = getattr(django_settings, "SERVESTATIC_KEEP_ONLY_HASHED_FILES", False)
         force_script_name = getattr(settings, "FORCE_SCRIPT_NAME", None)
         static_url = getattr(settings, "STATIC_URL", None)
         root = getattr(settings, "SERVESTATIC_ROOT", None)
@@ -126,15 +120,9 @@ async def __call__(self, request):
         if static_file is not None:
             return await self.aserve(static_file, request)
 
-        if django_settings.DEBUG and request.path.startswith(
-            django_settings.STATIC_URL
-        ):
+        if django_settings.DEBUG and request.path.startswith(django_settings.STATIC_URL):
             current_finders = finders.get_finders()
-            app_dirs = [
-                storage.location
-                for finder in current_finders
-                for storage in finder.storages.values()
-            ]
+            app_dirs = [storage.location for finder in current_finders for storage in finder.storages.values()]
             app_dirs = "\n• ".join(sorted(app_dirs))
             msg = f"ServeStatic did not find the file '{request.path.lstrip(django_settings.STATIC_URL)}' within the following paths:\n• {app_dirs}"
             raise MissingFileError(msg)
@@ -178,10 +166,7 @@ def add_files_from_finders(self):
 
     def add_files_from_manifest(self):
         if not isinstance(staticfiles_storage, ManifestStaticFilesStorage):
-            msg = (
-                "SERVESTATIC_USE_MANIFEST is set to True but "
-                "staticfiles storage is not using a manifest."
-            )
+            msg = "SERVESTATIC_USE_MANIFEST is set to True but " "staticfiles storage is not using a manifest."
             raise ValueError(msg)
         staticfiles: dict = staticfiles_storage.hashed_files
         stat_cache = None
@@ -190,10 +175,7 @@ def add_files_from_manifest(self):
         if hasattr(staticfiles_storage, "load_manifest_stats"):
             manifest_stats: dict = staticfiles_storage.load_manifest_stats()
             if manifest_stats:
-                stat_cache = {
-                    staticfiles_storage.path(k): os.stat_result(v)
-                    for k, v in manifest_stats.items()
-                }
+                stat_cache = {staticfiles_storage.path(k): os.stat_result(v) for k, v in manifest_stats.items()}
 
         # Add files to ServeStatic
         for unhashed_name, hashed_name in staticfiles.items():

src/servestatic/responders.py

@@ -146,9 +146,7 @@ async def aget_response(self, method, request_headers):
             # just ignore it and return the standard response (this
             # behaviour is allowed by the spec)
             with contextlib.suppress(ValueError):
-                return await self.aget_range_response(
-                    range_header, headers, file_handle
-                )
+                return await self.aget_range_response(range_header, headers, file_handle)
         return Response(HTTPStatus.OK, headers, file_handle)
 
     def get_range_response(self, range_header, base_headers, file_handle):
@@ -270,12 +268,8 @@ def get_headers(self, headers_list, files):
 
     @staticmethod
     def get_not_modified_response(headers):
-        not_modified_headers = [
-            (key, headers[key]) for key in NOT_MODIFIED_HEADERS if key in headers
-        ]
-        return Response(
-            status=HTTPStatus.NOT_MODIFIED, headers=not_modified_headers, file=None
-        )
+        not_modified_headers = [(key, headers[key]) for key in NOT_MODIFIED_HEADERS if key in headers]
+        return Response(status=HTTPStatus.NOT_MODIFIED, headers=not_modified_headers, file=None)
 
     @staticmethod
     def get_alternatives(base_headers, files):

src/servestatic/runserver_nostatic/management/commands/runserver.py

@@ -48,7 +48,4 @@ def add_arguments(self, parser):
         super().add_arguments(parser)
         if parser.get_default("use_static_handler") is True:
             parser.set_defaults(use_static_handler=False)
-            parser.description += (
-                "\n(Wrapped by 'servestatic.runserver_nostatic' to always"
-                " enable '--nostatic')"
-            )
+            parser.description += "\n(Wrapped by 'servestatic.runserver_nostatic' to always" " enable '--nostatic')"

src/servestatic/storage.py

@@ -30,9 +30,7 @@ class CompressedStaticFilesStorage(StaticFilesStorage):
 
     compressor: Compressor | None
 
-    def post_process(
-        self, paths: dict[str, Any], dry_run: bool = False, **options: Any
-    ) -> _PostProcessT:
+    def post_process(self, paths: dict[str, Any], dry_run: bool = False, **options: Any) -> _PostProcessT:
         if dry_run:
             return
 
@@ -41,9 +39,7 @@ def post_process(
 
         to_compress = (path for path in paths if self.compressor.should_compress(path))
         with concurrent.futures.ThreadPoolExecutor() as executor:
-            futures = (
-                executor.submit(self._compress_one, path) for path in to_compress
-            )
+            futures = (executor.submit(self._compress_one, path) for path in to_compress)
             for compressed_paths in concurrent.futures.as_completed(futures):
                 yield from compressed_paths.result()
 
@@ -52,8 +48,7 @@ def _compress_one(self, path: str) -> list[tuple[str, str, bool]]:
         full_path = self.path(path)
         prefix_len = len(full_path) - len(path)
         compressed.extend(
-            (path, compressed_path[prefix_len:], True)
-            for compressed_path in self.compressor.compress(full_path)
+            (path, compressed_path[prefix_len:], True) for compressed_path in self.compressor.compress(full_path)
         )
         return compressed
 
@@ -119,9 +114,7 @@ def stat_static_root(self):
 
         file_paths = []
         for root, _, files in os.walk(static_root):
-            file_paths.extend(
-                os.path.join(root, f) for f in files if f != self.manifest_name
-            )
+            file_paths.extend(os.path.join(root, f) for f in files if f != self.manifest_name)
         stats = stat_files(file_paths)
 
         # Remove the static root folder from the path
@@ -197,20 +190,15 @@ def compress_files(self, names):
 
         to_compress = (name for name in names if self.compressor.should_compress(name))
         with concurrent.futures.ThreadPoolExecutor() as executor:
-            futures = (
-                executor.submit(self._compress_one, name) for name in to_compress
-            )
+            futures = (executor.submit(self._compress_one, name) for name in to_compress)
             for compressed_paths in concurrent.futures.as_completed(futures):
                 yield from compressed_paths.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)
-        )
+        compressed.extend((name, compressed_path[prefix_len:]) for compressed_path in self.compressor.compress(path))
         return compressed
 
     def make_helpful_exception(self, exception, name):

src/servestatic/utils.py

@@ -68,18 +68,14 @@ def __init__(self, iterator: AsyncIterable):
     def __iter__(self):
         # Create a dedicated event loop to run the async iterator on.
         loop = asyncio.new_event_loop()
-        thread_executor = concurrent.futures.ThreadPoolExecutor(
-            max_workers=1, thread_name_prefix="ServeStatic"
-        )
+        thread_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="ServeStatic")
 
         # Convert from async to sync by stepping through the async iterator and yielding
         # the result of each step.
         generator = self.iterator.__aiter__()
         with contextlib.suppress(GeneratorExit, StopAsyncIteration):
             while True:
-                yield thread_executor.submit(
-                    loop.run_until_complete, generator.__anext__()
-                ).result()
+                yield thread_executor.submit(loop.run_until_complete, generator.__anext__()).result()
         loop.close()
         thread_executor.shutdown(wait=True)
 
@@ -129,9 +125,7 @@ def __init__(
             opener,
         )
         self.loop: asyncio.AbstractEventLoop | None = None
-        self.executor = ThreadPoolExecutor(
-            max_workers=1, thread_name_prefix="ServeStatic-AsyncFile"
-        )
+        self.executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ServeStatic-AsyncFile")
         self.lock = threading.Lock()
         self.file_obj: None | IOBase = None
         self.closed = False

tests/test_asgi.py

@@ -35,9 +35,7 @@ async def asgi_app(scope, receive, send):
         )
         await send({"type": "http.response.body", "body": b"Not Found"})
 
-    return ServeStaticASGI(
-        asgi_app, root=test_files.directory, autorefresh=request.param
-    )
+    return ServeStaticASGI(asgi_app, root=test_files.directory, autorefresh=request.param)
 
 
 def test_get_js_static_file(application, test_files):
@@ -93,19 +91,15 @@ def test_small_block_size(application, test_files):
 
 
 def test_request_range_response(application, test_files):
-    scope = AsgiScopeEmulator(
-        {"path": "/static/app.js", "headers": [(b"range", b"bytes=0-13")]}
-    )
+    scope = AsgiScopeEmulator({"path": "/static/app.js", "headers": [(b"range", b"bytes=0-13")]})
     receive = AsgiReceiveEmulator()
     send = AsgiSendEmulator()
     asyncio.run(application(scope, receive, send))
     assert send.body == test_files.js_content[:14]
 
 
 def test_out_of_range_error(application, test_files):
-    scope = AsgiScopeEmulator(
-        {"path": "/static/app.js", "headers": [(b"range", b"bytes=10000-11000")]}
-    )
+    scope = AsgiScopeEmulator({"path": "/static/app.js", "headers": [(b"range", b"bytes=10000-11000")]})
     receive = AsgiReceiveEmulator()
     send = AsgiSendEmulator()
     asyncio.run(application(scope, receive, send))

tests/test_compress.py

@@ -36,9 +36,7 @@ def files_dir():
 
 
 def test_compresses_file(files_dir):
-    with contextlib.closing(
-        gzip.open(os.path.join(files_dir, f"{COMPRESSABLE_FILE}.gz"), "rb")
-    ) as f:
+    with contextlib.closing(gzip.open(os.path.join(files_dir, f"{COMPRESSABLE_FILE}.gz"), "rb")) as f:
         contents = f.read()
     assert TEST_FILES[COMPRESSABLE_FILE] == contents
 
@@ -79,6 +77,4 @@ def test_compress():
 
 def test_compressed_effectively_no_orig_size():
     compressor = Compressor(quiet=True)
-    assert not compressor.is_compressed_effectively(
-        "test_encoding", "test_path", 0, "test_data"
-    )
+    assert not compressor.is_compressed_effectively("test_encoding", "test_path", 0, "test_data")