Enable mypy check (#1376)

Signed-off-by: Bala.FA <[email protected]>

Author: balamurugana

Makefile

@@ -3,14 +3,15 @@ default: tests
 
 getdeps:
 	@echo "Installing required dependencies"
-	@pip install --user --upgrade autopep8 certifi pytest pylint urllib3 argon2-cffi pycryptodome typing-extensions
+	@pip install --user --upgrade autopep8 certifi pytest pylint urllib3 argon2-cffi pycryptodome typing-extensions mypy
 
 check: getdeps
 	@echo "Running checks"
 	@pylint --reports=no --score=no --disable=R0401,R0801 minio/*py
 	@pylint --reports=no --score=no minio/credentials tests/functional
 	@isort --diff .
 	@find . -name "*.py" -exec autopep8 --diff --exit-code {} +
+	@mypy minio
 
 apply: getdeps
 	@isort .

minio/api.py

@@ -1224,7 +1224,7 @@ def get_object(
         check_non_empty_string(object_name)
         check_ssec(ssec)
 
-        headers = ssec.headers() if ssec else {}
+        headers = cast(DictType, ssec.headers() if ssec else {})
         headers.update(request_headers or {})
 
         if offset or length:
@@ -3171,7 +3171,7 @@ def _list_multipart_uploads(
             "GET",
             bucket_name,
             query_params=cast(DictType, query_params),
-            headers=cast(DictType | None, extra_headers),
+            headers=cast(Union[DictType, None], extra_headers),
         )
         return ListMultipartUploadsResult(response)
 
@@ -3214,6 +3214,6 @@ def _list_parts(
             bucket_name,
             object_name=object_name,
             query_params=cast(DictType, query_params),
-            headers=cast(DictType | None, extra_headers),
+            headers=cast(Union[DictType, None], extra_headers),
         )
         return ListPartsResult(response)

minio/helpers.py

@@ -302,7 +302,10 @@ def md5sum_hash(data: str | bytes | None) -> str | None:
 
     # indicate md5 hashing algorithm is not used in a security context.
     # Refer https://bugs.python.org/issue9216 for more information.
-    hasher = hashlib.new("md5", usedforsecurity=False)
+    hasher = hashlib.new(  # type: ignore[call-arg]
+        "md5",
+        usedforsecurity=False,
+    )
     hasher.update(data.encode() if isinstance(data, str) else data)
     md5sum = base64.b64encode(hasher.digest())
     return md5sum.decode() if isinstance(md5sum, bytes) else md5sum

minio/minioadmin.py

@@ -24,7 +24,7 @@
 import os
 from datetime import timedelta
 from enum import Enum
-from typing import TextIO
+from typing import TextIO, Tuple, cast
 from urllib.parse import urlunsplit
 
 import certifi
@@ -40,7 +40,7 @@
 from .crypto import decrypt, encrypt
 from .datatypes import PeerInfo, PeerSite, SiteReplicationStatusOptions
 from .error import MinioAdminException
-from .helpers import (_DEFAULT_USER_AGENT, _REGION_REGEX, _parse_url,
+from .helpers import (_DEFAULT_USER_AGENT, _REGION_REGEX, DictType, _parse_url,
                       headers_to_strings, queryencode, sha256_hash,
                       url_replace)
 from .signer import sign_v4_s3
@@ -149,8 +149,7 @@ def _url_open(
             self,
             method: str,
             command: CommandType,
-            query_params:
-                dict[str, str | list[str] | tuple[str]] | None = None,
+            query_params: DictType | None = None,
             body: bytes | None = None,
             preload_content: bool = True,
     ) -> BaseHTTPResponse:
@@ -169,7 +168,7 @@ def _url_open(
 
         content_sha256 = sha256_hash(body)
         date = time.utcnow()
-        headers: dict[str, str | list[str] | tuple[str]] = {
+        headers: DictType = {
             "Host": url.netloc,
             "User-Agent": self._user_agent,
             "x-amz-date": time.to_amz_date(date),
@@ -239,7 +238,7 @@ def _url_open(
         if response.status in [200, 204, 206]:
             return response
 
-        raise MinioAdminException(response.status, response.data.decode())
+        raise MinioAdminException(str(response.status), response.data.decode())
 
     def set_app_info(self, app_name: str, app_version: str):
         """
@@ -472,7 +471,7 @@ def policy_set(
             response = self._url_open(
                 "PUT",
                 _COMMAND.SET_USER_OR_GROUP_POLICY,
-                query_params={"userOrGroup": user or group,
+                query_params={"userOrGroup": cast(str, user or group),
                               "isGroup": "true" if group else "false",
                               "policyName": policy_name},
             )
@@ -585,7 +584,10 @@ def config_restore(self, restore_id: str) -> str:
         )
         return response.data.decode()
 
-    def profile_start(self, profilers: tuple[str] = ()) -> str:
+    def profile_start(
+            self,
+            profilers: tuple[str] = cast(Tuple[str], ()),
+    ) -> str:
         """Runs a system profile"""
         response = self._url_open(
             "POST",
@@ -607,7 +609,7 @@ def kms_key_create(self, key: str | None = None) -> str:
         response = self._url_open(
             "POST",
             _COMMAND.CREATE_KMS_KEY,
-            query_params={"key-id": key},
+            query_params={"key-id": key or ""},
         )
         return response.data.decode()
 
@@ -645,7 +647,7 @@ def get_site_replication_status(
         response = self._url_open(
             "GET",
             _COMMAND.SITE_REPLICATION_STATUS,
-            query_params=options.to_query_params(),
+            query_params=cast(DictType, options.to_query_params()),
         )
         return response.data.decode()
 
@@ -668,9 +670,9 @@ def remove_site_replication(
         """Remove given sites or all sites from site replication."""
         data = {}
         if all_sites:
-            data.update({"all": True})
+            data.update({"all": "True"})
         elif sites:
-            data.update({"sites": sites})
+            data.update({"sites": sites or ""})
         else:
             raise ValueError("either sites or all flag must be given")
         body = json.dumps(data).encode()

minio/py.typed

@@ -22,7 +22,7 @@
 from datetime import datetime, timezone
 
 try:
-    from datetime import UTC
+    from datetime import UTC  # type: ignore[attr-defined]
     _UTC_IMPORTED = True
 except ImportError:
     _UTC_IMPORTED = False