add stricter type checking

minio_storage/files.py

@@ -8,7 +8,7 @@
 from minio_storage.errors import minio_error
 
 if T.TYPE_CHECKING:
-    from minio_storage.storage import Storage
+    from minio_storage.storage import MinioStorage
 
 logger = getLogger("minio_storage")
 
@@ -36,8 +36,8 @@ def seek(self, *args, **kwargs) -> bool:
 
 
 class MinioStorageFile(File):
-    def __init__(self, name: str, mode: str, storage: "Storage", **kwargs):
-        self._storage: "Storage" = storage
+    def __init__(self, name: str, mode: str, storage: "MinioStorage", **kwargs):
+        self._storage: "MinioStorage" = storage
         self.name: str = name
         self._mode: str = mode
         self._file = None
@@ -52,7 +52,7 @@ def __init__(
         self,
         name: str,
         mode: str,
-        storage: "Storage",
+        storage: "MinioStorage",
         max_memory_size: T.Optional[int] = None,
         **kwargs,
     ):
@@ -66,6 +66,7 @@ def __init__(
 
     @property
     def file(self):
+        obj = None
         if self._file is None:
             try:
                 obj = self._storage.client.get_object(
@@ -78,7 +79,8 @@ def file(self):
                 raise OSError(f"File {self.name} does not exist")
             finally:
                 try:
-                    obj.release_conn()
+                    if obj:
+                        obj.release_conn()
                 except Exception as e:
                     logger.error(str(e))
         return self._file
@@ -104,7 +106,7 @@ def __init__(
         self,
         name: str,
         mode: str,
-        storage: "Storage",
+        storage: "MinioStorage",
         max_memory_size: T.Optional[int] = None,
         **kwargs,
     ):
@@ -119,6 +121,7 @@ def __init__(
     @property
     def file(self):
         if self._file is None:
+            obj = None
             try:
                 obj = self._storage.client.get_object(
                     self._storage.bucket_name, self.name
@@ -134,7 +137,8 @@ def file(self):
                 raise minio_error(f"File {self.name} does not exist", error)
             finally:
                 try:
-                    obj.release_conn()
+                    if obj:
+                        obj.release_conn()
                 except Exception as e:
                     logger.error(str(e))
         return self._file

minio_storage/management/commands/minio.py

@@ -1,4 +1,3 @@
-import argparse
 import json
 import sys
 import typing as T
@@ -46,12 +45,8 @@ def add_arguments(self, parser):
             description="valid subcommands",
             # required=True,
         )
-        cmds._parser_class = argparse.ArgumentParser  # circumvent Django 1.11 bug
-
         cmds.add_parser(self.CHECK, help="check bucket")
-
         cmds.add_parser(self.CREATE, help="make bucket")
-
         cmds.add_parser(self.DELETE, help="remove an empty bucket")
 
         ls = cmds.add_parser(self.LIST, help="list bucket objects or buckets")
@@ -141,7 +136,7 @@ def storage(self, options):
 
         # TODO: maybe another way
         with patch.object(storage_class, "_init_check", return_value=None):
-            storage = storage_class()
+            storage = storage_class()  # type: ignore
             return storage
 
     def bucket_exists(self, storage, bucket_name):

minio_storage/storage.py

@@ -12,6 +12,7 @@
 from django.core.files.storage import Storage
 from django.utils import timezone
 from django.utils.deconstruct import deconstructible
+from minio.datatypes import Object
 
 from minio_storage.errors import minio_error
 from minio_storage.files import ReadOnlySpooledTemporaryFile
@@ -106,12 +107,9 @@ def _create_base_url_client(client: minio.Minio, bucket_name: str, base_url: str
         base_url_parts = urlsplit(base_url)
 
         # Clone from the normal client, but with base_url as the endpoint
-        credentials = client._provider.retrieve()
         base_url_client = minio.Minio(
             base_url_parts.netloc,
-            access_key=credentials.access_key,
-            secret_key=credentials.secret_key,
-            session_token=credentials.session_token,
+            credentials=client._provider,
             secure=base_url_parts.scheme == "https",
             # The bucket region may be auto-detected by client (via an HTTP
             # request), so don't just use client._region
@@ -250,8 +248,8 @@ def listdir(self, path: str) -> T.Tuple[T.List, T.List]:
 
     def size(self, name: str) -> int:
         try:
-            info = self.client.stat_object(self.bucket_name, name)
-            return info.size
+            info: Object = self.client.stat_object(self.bucket_name, name)
+            return info.size  # type: ignore
         except merr.InvalidResponseError as error:
             raise minio_error(f"Could not access file size for {name}", error)
 
@@ -293,7 +291,7 @@ def _presigned_url(
 
     def url(
         self, name: str, *args, max_age: T.Optional[datetime.timedelta] = None
-    ) -> T.Optional[str]:
+    ) -> str:
         url = ""
         if self.presign_urls:
             url = self._presigned_url(name, max_age=max_age)
@@ -317,7 +315,9 @@ def strip_end(path):
                     self.bucket_name,
                     quote(strip_beg(name)),
                 )
-        return url
+        if url:
+            return url
+        raise OSError(f"could not produce URL for {name}")
 
     @property
     def endpoint_url(self):
@@ -337,12 +337,14 @@ def created_time(self, name: str) -> datetime.datetime:
 
     def modified_time(self, name: str) -> datetime.datetime:
         try:
-            info = self.client.stat_object(self.bucket_name, name)
-            return info.last_modified
+            info: Object = self.client.stat_object(self.bucket_name, name)
+            if info.last_modified:
+                return info.last_modified  # type: ignore
         except merr.InvalidResponseError as error:
             raise minio_error(
                 f"Could not access modification time for file {name}", error
             )
+        raise OSError(f"Could not access modification time for file {name}")
 
 
 _NoValue = object()

tests/test_app/tests/custom_storage_class_tests.py

@@ -96,14 +96,15 @@ def test_custom_storage(self):
             ) as storage_file:
                 # Copy the stream from the http stream to the out_file
                 #
-                shutil.copyfileobj(storage_file.file, out_file)
+                shutil.copyfileobj(storage_file.file, out_file)  # type: ignore
 
                 #
                 # We are not using the ReadOnlyMinioObjectFile type so we can't seek in
                 # it.
                 #
                 with self.assertRaises(io.UnsupportedOperation):
-                    storage_file.file.seek()
+                    if storage_file.file:
+                        storage_file.file.seek(0)
 
             workspace_files = os.listdir(workspace)
             print(workspace_files)  # prints: ['secret.txt']

tests/test_app/tests/upload_tests.py

@@ -92,7 +92,7 @@ def test_metadata(self):
             "X-Xss-Protection",
             "Date",
         }
-        self.assertTrue(metadata_attrs.issubset(res.metadata.keys()))
+        self.assertTrue(metadata_attrs.issubset(res.metadata.keys()))  # type: ignore
 
 
 @override_settings(
@@ -105,4 +105,4 @@ def test_default_metadata(self):
             self.media_storage.bucket_name, ivan
         )
 
-        self.assertEqual(res.metadata["Cache-Control"], "max-age=1000")
+        self.assertEqual(res.metadata["Cache-Control"], "max-age=1000")  # type: ignore

tox.ini

@@ -55,7 +55,7 @@ deps =
         django-stubs==4.2.0
         -rdev-requirements.txt
 commands =
-    pyright --skipunannotated --level WARNING
+    pyright --level WARNING
 
 
 [testenv:lint]