remove type hint ignores

.gitignore

@@ -12,3 +12,4 @@ storage-credentials
 .mypy_cache
 .idea
 .python-version
+.dmypy.json

.pre-commit-config.yaml

@@ -62,6 +62,8 @@ repos:
           - django-storages
           - boto3
           - google-cloud-storage
+          - boto3-stubs[s3]
+
   - repo: https://github.com/mgedmin/check-manifest
     rev: "0.49"
     hooks:

CHANGELOG.md

@@ -1,4 +1,7 @@
 # Changelog
+## 3.1.1
+- removed type ignores, updated tests
+
 ## 3.1.0
 - add new strategies for two-pass collectstatic where the first pass is file or memory based
 

collectfasta/strategies/boto3.py

@@ -1,13 +1,17 @@
-# type: ignore
 import logging
 import os
 import tempfile
 from typing import Any
+from typing import Dict
 from typing import Optional
+from typing import TypeVar
+from typing import Union
+from typing import cast
 
 import botocore.exceptions
 from boto3.s3.transfer import TransferConfig
 from django.contrib.staticfiles.storage import ManifestFilesMixin
+from django.core.files.storage import Storage
 from django.utils.timezone import make_naive
 from storages.backends.s3boto3 import S3Boto3Storage
 from storages.backends.s3boto3 import S3ManifestStaticStorage
@@ -27,11 +31,12 @@
 logger = logging.getLogger(__name__)
 
 
-class S3StorageWrapperMixin:
-    def __init__(self, original: S3Boto3Storage) -> None:
+class S3StorageWrapperBase(S3Boto3Storage):
+    def __init__(self, *args: Any, original: Any, **kwargs: Any) -> None:
         default_settings = original.get_default_settings()
         for name in default_settings.keys():
-            if not hasattr(self, name) and hasattr(original, name):
+            setattr(self, name, default_settings[name])
+            if hasattr(original, name):
                 setattr(self, name, getattr(original, name))
         for arg in [
             "_bucket",
@@ -49,32 +54,29 @@ def __init__(self, original: S3Boto3Storage) -> None:
             self._transfer_config = TransferConfig(use_threads=self.use_threads)
 
         self.preload_metadata = True
-        self._entries = {}
+        self._entries: Dict[str, str] = {}
 
     # restores the "preload_metadata" method that was removed in django-storages 1.10
     def _save(self, name, content):
-        content.seek(0)
-        with tempfile.SpooledTemporaryFile() as tmp:
-            tmp.write(content.read())
-            cleaned_name = clean_name(name)
-            name = self._normalize_name(cleaned_name)
-            params = self._get_write_parameters(name, content)
-
-            if is_seekable(content):
-                content.seek(0, os.SEEK_SET)
-            if (
-                self.gzip
-                and params["ContentType"] in self.gzip_content_types
-                and "ContentEncoding" not in params
-            ):
-                content = self._compress_content(content)
-                params["ContentEncoding"] = "gzip"
-
-            obj = self.bucket.Object(name)
-            if self.preload_metadata:
-                self._entries[name] = obj
-            obj.upload_fileobj(content, ExtraArgs=params, Config=self._transfer_config)
-            return cleaned_name
+        cleaned_name = clean_name(name)
+        name = self._normalize_name(cleaned_name)
+        params = self._get_write_parameters(name, content)
+
+        if is_seekable(content):
+            content.seek(0, os.SEEK_SET)
+        if (
+            self.gzip
+            and params["ContentType"] in self.gzip_content_types
+            and "ContentEncoding" not in params
+        ):
+            content = self._compress_content(content)
+            params["ContentEncoding"] = "gzip"
+
+        obj = self.bucket.Object(name)
+        if self.preload_metadata:
+            self._entries[name] = obj
+        obj.upload_fileobj(content, ExtraArgs=params, Config=self._transfer_config)
+        return cleaned_name
 
     @property
     def entries(self):
@@ -121,11 +123,11 @@ def get_modified_time(self, name):
             return make_naive(entry.last_modified)
 
 
-class ManifestFilesWrapperMixin:
-    def __init__(self, original: ManifestFilesMixin) -> None:
-        super().__init__(original)
+class ManifestFilesWrapper(ManifestFilesMixin):
+    def __init__(self, *args: Any, original: Any, **kwargs: Any) -> None:
+        super().__init__(*args, original=original, **kwargs)
         if original.manifest_storage == original:
-            self.manifest_storage = self
+            self.manifest_storage = cast(Storage, self)
         else:
             self.manifest_storage = original.manifest_storage
         for arg in [
@@ -140,37 +142,51 @@ def __init__(self, original: ManifestFilesMixin) -> None:
                 setattr(self, arg, getattr(original, arg))
 
 
-class S3StorageWrapper(S3StorageWrapperMixin, S3Boto3Storage):
+class S3StorageWrapper(S3StorageWrapperBase, S3Boto3Storage):
     pass
 
 
-class S3StaticStorageWrapper(S3StorageWrapperMixin, S3StaticStorage):
+class S3StaticStorageWrapper(S3StorageWrapperBase, S3StaticStorage):
     pass
 
 
 class S3ManifestStaticStorageWrapper(
-    ManifestFilesWrapperMixin, S3StorageWrapperMixin, S3ManifestStaticStorage
+    ManifestFilesWrapper,
+    S3StorageWrapperBase,
+    S3ManifestStaticStorage,
 ):
-    pass
+    def _save(self, name, content):
+        content.seek(0)
+        with tempfile.SpooledTemporaryFile() as tmp:
+            tmp.write(content.read())
+            return super()._save(name, tmp)
+
+
+S3Storage = TypeVar(
+    "S3Storage", bound=Union[S3Boto3Storage, S3StaticStorage, S3ManifestStaticStorage]
+)
+
+S3StorageWrapped = Union[
+    S3StaticStorageWrapper, S3ManifestStaticStorageWrapper, S3StorageWrapper
+]
 
 
-class Boto3Strategy(CachingHashStrategy[S3Boto3Storage]):
-    def __init__(self, remote_storage: S3Boto3Storage) -> None:
+class Boto3Strategy(CachingHashStrategy[S3Storage]):
+    def __init__(self, remote_storage: S3Storage) -> None:
         self.remote_storage = self.wrapped_storage(remote_storage)
         super().__init__(self.remote_storage)
         self.use_gzip = settings.aws_is_gzipped
-        self._entries: Optional[dict[str, Any]] = None
 
-    def wrapped_storage(self, remote_storage: S3Boto3Storage) -> S3Boto3Storage:
+    def wrapped_storage(self, remote_storage: S3Storage) -> S3StorageWrapped:
         if isinstance(remote_storage, S3ManifestStaticStorage):
-            return S3ManifestStaticStorageWrapper(remote_storage)
+            return S3ManifestStaticStorageWrapper(original=remote_storage)
         elif isinstance(remote_storage, S3StaticStorage):
-            return S3StaticStorageWrapper(remote_storage)
+            return S3StaticStorageWrapper(original=remote_storage)
         elif isinstance(remote_storage, S3Boto3Storage):
-            return S3StorageWrapper(remote_storage)
+            return S3StorageWrapper(original=remote_storage)
         return remote_storage
 
-    def wrap_storage(self, remote_storage: S3Boto3Storage) -> S3Boto3Storage:
+    def wrap_storage(self, remote_storage: S3Storage) -> S3StorageWrapped:
         return self.remote_storage
 
     def _normalize_path(self, prefixed_path: str) -> str:

collectfasta/strategies/hashing.py

@@ -1,5 +1,12 @@
+from typing import Any
+from typing import Callable
 from typing import Optional
+from typing import Protocol
+from typing import Tuple
 from typing import Type
+from typing import Union
+from typing import cast
+from typing import runtime_checkable
 
 from django.contrib.staticfiles.storage import ManifestFilesMixin
 from django.core.files.storage import FileSystemStorage
@@ -8,15 +15,30 @@
 
 from .base import HashStrategy
 from .base import Strategy
-from .base import _RemoteStorage
 
 
-class InMemoryManifestFilesStorage(ManifestFilesMixin, InMemoryStorage):  # type: ignore
-    pass
+@runtime_checkable
+class LocationConstructorProtocol(Protocol):
 
+    def __init__(self, location: Optional[str]) -> None: ...
 
-class FileSystemManifestFilesStorage(ManifestFilesMixin, FileSystemStorage):  # type: ignore
-    pass
+
+@runtime_checkable
+class HasLocationProtocol(Protocol):
+    location: str
+
+
+class InMemoryManifestFilesStorage(ManifestFilesMixin, InMemoryStorage):
+    url: Callable[..., Any]
+
+
+class FileSystemManifestFilesStorage(ManifestFilesMixin, FileSystemStorage):
+    url: Callable[..., Any]
+
+
+OriginalStorage = Union[
+    LocationConstructorProtocol, HasLocationProtocol, Storage, ManifestFilesMixin
+]
 
 
 class HashingTwoPassStrategy(HashStrategy[Storage]):
@@ -26,23 +48,28 @@ class HashingTwoPassStrategy(HashStrategy[Storage]):
     the files to the remote storage
     """
 
-    first_manifest_storage: Optional[Type[ManifestFilesMixin]] = None
-    second_strategy: Optional[Type[Strategy]] = None
+    first_manifest_storage: Type[OriginalStorage]
+    second_strategy: Type[Strategy[Storage]]
+    original_storage: OriginalStorage
+    memory_storage: OriginalStorage
 
-    def __init__(self, remote_storage: _RemoteStorage) -> None:
+    def __init__(self, remote_storage: OriginalStorage) -> None:
+        assert issubclass(self.first_manifest_storage, ManifestFilesMixin)
+        assert isinstance(remote_storage, ManifestFilesMixin)
         self.first_pass = True
         self.original_storage = remote_storage
         self.memory_storage = self._get_tmp_storage()
+        assert isinstance(self.memory_storage, Storage)
         self.remote_storage = self.memory_storage
         super().__init__(self.memory_storage)
 
-    def _get_tmp_storage(self) -> Storage:
-        if isinstance(self.original_storage, ManifestFilesMixin):
-            return self.first_manifest_storage(location=self.original_storage.location)  # type: ignore
-        else:
-            raise ValueError(
-                "HashingMemoryStrategy can only be used with subclasses of ManifestFilesMixin"
-            )
+    def _get_tmp_storage(self) -> OriginalStorage:
+        assert isinstance(self.original_storage, HasLocationProtocol)
+        assert issubclass(self.first_manifest_storage, LocationConstructorProtocol)
+        return cast(
+            OriginalStorage,
+            self.first_manifest_storage(location=self.original_storage.location),
+        )
 
     def wrap_storage(self, remote_storage: Storage) -> Storage:
         return self.remote_storage
@@ -63,14 +90,24 @@ def second_pass_strategy(self):
                 "second_strategy must be set to a valid strategy class"
             )
         else:
+            assert isinstance(self.original_storage, Storage)
             return self.second_strategy(self.original_storage)
 
 
-class WithoutPrefixMixin:
-    def copy_args_hook(self, args):
+Task = Tuple[str, str, Storage]
+
+
+class StrategyWithLocationProtocol:
+    remote_storage: Any
+
+
+class WithoutPrefixMixin(StrategyWithLocationProtocol):
+
+    def copy_args_hook(self, args: Task) -> Task:
+        assert isinstance(self.remote_storage, HasLocationProtocol)
         return (
-            args[0].replace(self.remote_storage.location, ""),  # type: ignore
-            args[1].replace(self.remote_storage.location, ""),  # type: ignore
+            args[0].replace(self.remote_storage.location, ""),
+            args[1].replace(self.remote_storage.location, ""),
             args[2],
         )
 

collectfasta/tests/command/test_command.py

@@ -1,6 +1,5 @@
 import timeit
 from typing import Callable
-from typing import ContextManager
 from typing import Iterator
 from unittest import TestCase
 from unittest import mock
@@ -27,7 +26,7 @@
 
 def generate_storage_variations(
     storages: dict, strategies: dict
-) -> Iterator[tuple[str, ContextManager]]:
+) -> Iterator[tuple[str, override_settings]]:
     for storage_name, storage_settings in storages.items():
         for strategy_name, strategy_settings in strategies.items():
             yield (
@@ -102,9 +101,9 @@ def generate_storage_variations(
     **filesystem_backend_confs,
 }
 
-make_test_aws_backends: Callable = many(**aws_backend_confs)  # type: ignore
-make_test_all_backends: Callable = many(**all_backend_confs)  # type: ignore
-make_test_cloud_backends: Callable = many(**aws_backend_confs, **gcloud_backend_confs)  # type: ignore
+make_test_aws_backends: Callable = many(**aws_backend_confs)
+make_test_all_backends: Callable = many(**all_backend_confs)
+make_test_cloud_backends: Callable = many(**aws_backend_confs, **gcloud_backend_confs)
 
 
 @make_test_all_backends
@@ -117,6 +116,16 @@ def test_basics(case: TestCase) -> None:
     case.assertIn("0 static files copied.", call_collectstatic())
 
 
+@make_test_all_backends
+@live_test
+def test_only_copies_new(case: TestCase) -> None:
+    clean_static_dir()
+    create_two_referenced_static_files()
+    assert_static_file_number(2, call_collectstatic(), case)
+    create_two_referenced_static_files()
+    assert_static_file_number(2, call_collectstatic(), case)
+
+
 @make_test_all_backends
 @live_test
 @override_setting("threads", 5)