Type annotations on plain/assets

Author: davegaeddert

plain-admin/plain/admin/views/types.py

@@ -6,7 +6,7 @@ def __init__(
         alt: str = "",
         width: int | None = None,
         height: int | None = None,
-    ) -> None:
+    ):
         self.src = src
         self.alt = alt
         self.width = width

plain/plain/assets/compile.py

@@ -1,10 +1,14 @@
+from __future__ import annotations
+
 import gzip
 import os
 import shutil
+from collections.abc import Iterator
+from pathlib import Path
 
 from plain.runtime import PLAIN_TEMP_PATH
 
-from .finders import iter_assets
+from .finders import Asset, iter_assets
 from .fingerprints import AssetsFingerprintsManifest, get_file_fingerprint
 
 SKIP_COMPRESS_EXTENSIONS = (
@@ -40,7 +44,7 @@
 )
 
 
-def get_compiled_path():
+def get_compiled_path() -> Path:
     """
     Get the path at runtime to the compiled assets directory.
 
@@ -49,7 +53,9 @@ def get_compiled_path():
     return PLAIN_TEMP_PATH / "assets" / "compiled"
 
 
-def compile_assets(*, target_dir, keep_original, fingerprint, compress):
+def compile_assets(
+    *, target_dir: str, keep_original: bool, fingerprint: bool, compress: bool
+) -> Iterator[tuple[str, str, list[str]]]:
     """
     Compile all assets to the target directory and save a JSON manifest
     mapping the original filenames to the compiled filenames.
@@ -73,17 +79,24 @@ def compile_assets(*, target_dir, keep_original, fingerprint, compress):
     manifest.save()
 
 
-def compile_asset(*, asset, target_dir, keep_original, fingerprint, compress):
+def compile_asset(
+    *,
+    asset: Asset,
+    target_dir: str,
+    keep_original: bool,
+    fingerprint: bool,
+    compress: bool,
+) -> tuple[str, list[str]]:
     """
     Compile an asset to multiple output paths.
     """
-    compiled_paths = []
+    compiled_paths: list[str] = []
 
     # The expected destination for the original asset
     target_path = os.path.join(target_dir, asset.url_path)
 
     # Keep track of where the final, resolved asset ends up
-    resolved_url_path = asset.url_path
+    resolved_url_path: str = asset.url_path
 
     # Make sure all the expected directories exist
     os.makedirs(os.path.dirname(target_path), exist_ok=True)
@@ -106,7 +119,7 @@ def compile_asset(*, asset, target_dir, keep_original, fingerprint, compress):
         shutil.copy(asset.absolute_path, fingerprinted_path)
         compiled_paths.append(fingerprinted_path)
 
-        resolved_url_path = os.path.relpath(fingerprinted_path, target_dir)
+        resolved_url_path = str(os.path.relpath(fingerprinted_path, target_dir))
 
     if compress and extension.lower() not in SKIP_COMPRESS_EXTENSIONS:
         for path in compiled_paths.copy():

plain/plain/assets/finders.py

@@ -1,4 +1,7 @@
+from __future__ import annotations
+
 import os
+from collections.abc import Iterator
 
 from plain.packages import packages_registry
 from plain.runtime import APP_PATH
@@ -8,21 +11,22 @@
 SKIP_ASSETS = (".DS_Store", ".gitignore")
 
 
-def iter_assets():
+class Asset:
+    def __init__(self, *, url_path: str, absolute_path: str):
+        self.url_path = url_path
+        self.absolute_path = absolute_path
+
+    def __str__(self) -> str:
+        return self.url_path
+
+
+def iter_assets() -> Iterator[Asset]:
     """
     Iterate all valid asset files found in the installed
     packages and the app itself.
     """
 
-    class Asset:
-        def __init__(self, *, url_path, absolute_path):
-            self.url_path = url_path
-            self.absolute_path = absolute_path
-
-        def __str__(self):
-            return self.url_path
-
-    def _iter_assets_dir(path):
+    def _iter_assets_dir(path: str) -> Iterator[tuple[str, str]]:
         for root, _, files in os.walk(path):
             for f in files:
                 if f in SKIP_ASSETS:
@@ -36,7 +40,7 @@ def _iter_assets_dir(path):
             yield Asset(url_path=url_path, absolute_path=abs_path)
 
 
-def iter_asset_dirs():
+def iter_asset_dirs() -> Iterator[str]:
     """
     Iterate all directories containing assets, from installed
     packages and from app/assets.

plain/plain/assets/fingerprints.py

@@ -15,18 +15,18 @@ class AssetsFingerprintsManifest(dict):
     def __init__(self):
         self.path = PLAIN_TEMP_PATH / "assets" / "fingerprints.json"
 
-    def load(self):
+    def load(self) -> None:
         if self.path.exists():
             with open(self.path) as f:
                 self.update(json.load(f))
 
-    def save(self):
+    def save(self) -> None:
         with open(self.path, "w") as f:
             json.dump(self, f, indent=2)
 
 
 @cache
-def _get_manifest():
+def _get_manifest() -> AssetsFingerprintsManifest:
     """
     A cached function for loading the asset fingerprints manifest,
     so we don't have to keep loading it from disk over and over.
@@ -36,16 +36,17 @@ def _get_manifest():
     return manifest
 
 
-def get_fingerprinted_url_path(url_path):
+def get_fingerprinted_url_path(url_path: str) -> str | None:
     """
     Get the final fingerprinted path for an asset URL path.
     """
     manifest = _get_manifest()
     if url_path in manifest:
         return manifest[url_path]
+    return None
 
 
-def get_file_fingerprint(file_path):
+def get_file_fingerprint(file_path: str) -> str:
     """
     Get the fingerprint hash for a file.
     """

plain/plain/assets/urls.py

@@ -18,7 +18,7 @@ class AssetsRouter(Router):
     ]
 
 
-def get_asset_url(url_path):
+def get_asset_url(url_path: str) -> str:
     """
     Get the full URL to a given asset path.
     """

plain/plain/assets/views.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import functools
 import mimetypes
 import os
@@ -28,14 +30,14 @@ class AssetView(View):
     This class could be subclassed to further tweak the responses or behavior.
     """
 
-    def __init__(self, asset_path=None):
+    def __init__(self, asset_path: str | None = None):
         # Allow a path to be passed in AssetView.as_view(path="...")
         self.asset_path = asset_path
 
-    def get_url_path(self):
+    def get_url_path(self) -> str:
         return self.asset_path or self.url_kwargs["path"]
 
-    def get(self):
+    def get(self) -> Response | FileResponse:
         url_path = self.get_url_path()
 
         # Make a trailing slash work, but we don't expect it
@@ -71,7 +73,7 @@ def get(self):
         response.headers = self.update_headers(response.headers, absolute_path)
         return response
 
-    def get_asset_path(self, path):
+    def get_asset_path(self, path: str) -> str:
         """Get the path to the compiled asset"""
         compiled_path = os.path.abspath(get_compiled_path())
         asset_path = os.path.join(compiled_path, path)
@@ -82,13 +84,14 @@ def get_asset_path(self, path):
 
         return asset_path
 
-    def get_debug_asset_path(self, path):
+    def get_debug_asset_path(self, path: str) -> str | None:
         """Make a "live" check to find the uncompiled asset in the filesystem"""
         for asset in iter_assets():
             if asset.url_path == path:
                 return asset.absolute_path
+        return None
 
-    def check_asset_path(self, path):
+    def check_asset_path(self, path: str | None) -> None:
         if not path:
             raise Http404("Asset not found")
 
@@ -99,17 +102,18 @@ def check_asset_path(self, path):
             raise Http404("Asset is a directory")
 
     @functools.cache
-    def get_last_modified(self, path):
+    def get_last_modified(self, path: str) -> str | None:
         try:
             mtime = os.path.getmtime(path)
         except OSError:
             mtime = None
 
         if mtime:
             return formatdate(mtime, usegmt=True)
+        return None
 
     @functools.cache
-    def get_etag(self, path):
+    def get_etag(self, path: str) -> str:
         try:
             mtime = os.path.getmtime(path)
         except OSError:
@@ -120,10 +124,10 @@ def get_etag(self, path):
         return f'"{timestamp:x}-{size:x}"'
 
     @functools.cache
-    def get_size(self, path):
+    def get_size(self, path: str) -> int:
         return os.path.getsize(path)
 
-    def update_headers(self, headers, path):
+    def update_headers(self, headers: dict, path: str) -> dict:
         headers.setdefault("Access-Control-Allow-Origin", "*")
 
         # Always vary on Accept-Encoding
@@ -165,7 +169,7 @@ def update_headers(self, headers, path):
 
         return headers
 
-    def is_immutable(self, path):
+    def is_immutable(self, path: str) -> bool:
         """
         Determine whether an asset looks like it is immutable.
 
@@ -182,14 +186,14 @@ def is_immutable(self, path):
 
         return False
 
-    def get_encoded_path(self, path):
+    def get_encoded_path(self, path: str) -> str | None:
         """
         If the client supports compression, return the path to the compressed file.
         Otherwise, return the original path.
         """
         accept_encoding = self.request.headers.get("Accept-Encoding")
         if not accept_encoding:
-            return
+            return None
 
         if "br" in accept_encoding:
             br_path = path + ".br"
@@ -200,15 +204,16 @@ def get_encoded_path(self, path):
             gzip_path = path + ".gz"
             if os.path.exists(gzip_path):
                 return gzip_path
+        return None
 
-    def get_redirect_response(self, path):
+    def get_redirect_response(self, path: str) -> ResponseRedirect | None:
         """If the asset is not found, try to redirect to the fingerprinted path"""
         fingerprinted_url_path = get_fingerprinted_url_path(path)
 
         if not fingerprinted_url_path or fingerprinted_url_path == path:
             # Don't need to redirect if there is no fingerprinted path,
             # or we're already looking at it.
-            return
+            return None
 
         from .urls import AssetsRouter
 
@@ -221,7 +226,7 @@ def get_redirect_response(self, path):
             },
         )
 
-    def get_conditional_response(self, path):
+    def get_conditional_response(self, path: str) -> ResponseNotModified | None:
         """
         Support conditional requests (HTTP 304 response) based on ETag and Last-Modified headers.
         """
@@ -241,8 +246,9 @@ def get_conditional_response(self, path):
                 response = ResponseNotModified()
                 response.headers = self.update_headers(response.headers, path)
                 return response
+        return None
 
-    def get_range_response(self, path):
+    def get_range_response(self, path: str) -> Response | StreamingResponse | None:
         """
         Support range requests (HTTP 206 response).
         """