chore: Add mypy as a type checker for repository (#11)

- Type hint modernization using mypy as a type checker on pre-commit -- solving all typing issues in the library.
- Bug fix when doing a proxy assignment.
- Remove Python 3.9 from labels.
- Add pyupgrade and isort to the default ruff rules (for typing upgrades with Python 3.10).

Author: bvanelli

.pre-commit-config.yaml

@@ -7,6 +7,14 @@ repos:
           - '--fix'
           - '--exit-non-zero-on-fix'
       - id: ruff-format
+  - repo: 'https://github.com/pre-commit/mirrors-mypy'
+    rev: v1.20.0
+    hooks:
+      - id: mypy
+        additional_dependencies:
+          - aiohttp
+          - async-lru
+          - rich
   - repo: 'https://github.com/pre-commit/pre-commit-hooks'
     rev: v5.0.0
     hooks:

crpy/__init__.py

@@ -1,6 +1,6 @@
-from crpy.registry import RegistryInfo
+from crpy.common import BaseCrpyError, HTTPConnectionError, UnauthorizedError
 from crpy.image import Blob, Image
-from crpy.common import HTTPConnectionError, UnauthorizedError, BaseCrpyError
+from crpy.registry import RegistryInfo
 from crpy.version import __version__
 
 __all__ = [

crpy/auth.py

@@ -5,16 +5,16 @@
 
 async def get_token(
     url: str,
-    username: str = None,
-    password: str = None,
-    b64_token: str = None,
-    aiohttp_kwargs: dict = None,
+    username: str | None = None,
+    password: str | None = None,
+    b64_token: str | None = None,
+    aiohttp_kwargs: dict | None = None,
 ):
     # get the credentials here.
     # I'll use the simple auth since it mostly works
     headers = {}
     if username and password:
-        token = b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
+        token = b64encode(f"{username}:{password}".encode()).decode("ascii")
         headers = {"Authorization": f"Basic {token}"}
     elif b64_token:
         headers = {"Authorization": f"Basic {b64_token}"}

crpy/cmd.py

@@ -8,7 +8,6 @@
 from rich.table import Table
 from rich.text import Text
 
-
 from crpy.common import HTTPConnectionError, UnauthorizedError
 from crpy.registry import RegistryInfo
 from crpy.storage import (

crpy/common.py

@@ -4,8 +4,7 @@
 import io
 import json
 import socket
-from dataclasses import dataclass
-from typing import List, Optional, Union
+from dataclasses import dataclass, field
 
 import aiohttp
 
@@ -14,19 +13,19 @@
 class Response:
     status: int
     data: bytes
-    headers: Optional[dict] = None
+    headers: dict = field(default_factory=dict)
 
     def json(self) -> dict:
         return json.loads(self.data)
 
 
 async def _request(
     url,
-    headers: dict = None,
-    params: dict = None,
-    data: Union[dict, bytes] = None,
+    headers: dict | None = None,
+    params: dict | None = None,
+    data: dict | bytes | None = None,
     method: str = "post",
-    aiohttp_kwargs: dict = None,
+    aiohttp_kwargs: dict | None = None,
 ) -> Response:
     aiohttp_kwargs = aiohttp_kwargs or {}
     try:
@@ -38,15 +37,15 @@ async def _request(
         raise HTTPConnectionError(str(e))
 
 
-async def _stream(url, headers: dict = None, aiohttp_kwargs: dict = None):
+async def _stream(url, headers: dict | None = None, aiohttp_kwargs: dict | None = None):
     aiohttp_kwargs = aiohttp_kwargs or {}
     async with aiohttp.ClientSession(trust_env=True) as session:
         async with session.get(url, headers=headers, **aiohttp_kwargs) as response:
             async for data, _ in response.content.iter_chunks():
                 yield data
 
 
-def compute_sha256(file: Union[str, io.BytesIO, bytes], use_prefix: bool = True):
+def compute_sha256(file: str | io.BytesIO | bytes, use_prefix: bool = True):
     # If input is a string, consider it a filename
     if isinstance(file, str):
         with open(file, "rb") as f:
@@ -92,7 +91,7 @@ def architecture(self) -> str:
         return self.value.split("/")[1]
 
     @property
-    def variant(self) -> Optional[str]:
+    def variant(self) -> str | None:
         split_value = self.value.split("/")
         return split_value[2] if len(split_value) > 2 else None
 
@@ -104,7 +103,7 @@ def platform_from_dict(platform: dict) -> str:
     return base_str
 
 
-async def resolve_hostname(hostname: str, family: int = socket.AF_UNSPEC) -> List[str]:
+async def resolve_hostname(hostname: str, family: int = socket.AF_UNSPEC) -> list[str]:
     """
     Resolves a hostname to a sorted list of unique IP addresses.
 

crpy/image.py

@@ -4,24 +4,24 @@
 import pathlib
 import tarfile
 import tempfile
+from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import List, Optional, Union
 
 from crpy.common import compute_sha256
 
-INPUT_TYPES = Union[bytes, pathlib.Path, str, io.StringIO, dict, None]
-
 
 @dataclass
 class Blob:
-    path: Optional[pathlib.Path] = None
-    content: Optional[bytes] = None
-    filename: Optional[pathlib.Path] = None
-    digest: Optional[str] = None
+    path: pathlib.Path | None = None
+    content: bytes | None = None
+    filename: pathlib.Path | None = None
+    digest: str | None = None
 
     @classmethod
-    def from_any(cls, value: INPUT_TYPES, digest: str = None) -> "Blob":
-        if isinstance(value, str):
+    def from_any(cls, value: "INPUT_TYPES", digest: str | None = None) -> "Blob":
+        if isinstance(value, Blob):
+            return value
+        elif isinstance(value, str):
             return cls(pathlib.Path(value), digest=digest)
         elif isinstance(value, pathlib.Path):
             return cls(value, digest=digest)
@@ -31,12 +31,14 @@ def from_any(cls, value: INPUT_TYPES, digest: str = None) -> "Blob":
             return cls(None, value.read().encode(), digest=digest)
         elif isinstance(value, dict):
             return cls(None, json.dumps(value).encode(), digest=digest)
+        raise ValueError(f"Unsupported type {type(value)}")
 
-    def as_bytes(self):
+    def as_bytes(self) -> bytes:
         if self.path:
             return self.path.read_bytes()
-        else:
-            return self.content
+        if self.content is None:
+            raise ValueError("Blob has no path or content")
+        return self.content
 
     def as_dict(self):
         return json.loads(self.as_bytes())
@@ -48,19 +50,22 @@ def sha256_sum(self):
         return self.digest
 
 
+INPUT_TYPES = bytes | pathlib.Path | str | io.StringIO | dict | Blob | None
+
+
 class Image:
     """
     Component to interact with docker images for the purpose of building and generating tar-files with the correct
     layers. This can be populated at will and written to disk, having the individual blobs modified.
     """
 
-    def __init__(self, config: dict = None, manifest: dict = None, layers: List[bytes] = None):
-        self._config: Optional[Blob] = None
-        self._manifest: Optional[Blob] = None
-        self._layers: Optional[List[Blob]] = None
+    def __init__(self, config: INPUT_TYPES, manifest: INPUT_TYPES, layers: Sequence[INPUT_TYPES]):
+        self._config: Blob
+        self._manifest: Blob
+        self._layers: list[Blob]
         self.config = config
         self.manifest = manifest
-        self.layers = layers or []
+        self.layers = layers
 
     @property
     def manifest(self) -> Blob:
@@ -79,14 +84,14 @@ def config(self, value: INPUT_TYPES):
         self._config = Blob.from_any(value)
 
     @property
-    def layers(self) -> List[Blob]:
+    def layers(self) -> list[Blob]:
         return self._layers
 
     @layers.setter
-    def layers(self, layers: List[INPUT_TYPES]):
+    def layers(self, layers: Sequence[INPUT_TYPES]):
         self._layers = [Blob.from_any(layer) for layer in layers]
 
-    def to_disk(self, filename: pathlib.Path, tags: List[str] = None):
+    def to_disk(self, filename: pathlib.Path | io.BytesIO | str, tags: list[str] | None = None):
         with tempfile.TemporaryDirectory() as temp_dir:
             web_manifest = self.manifest.as_dict()
             config_filename = f"{web_manifest['config']['digest'].split(':')[1]}.json"
@@ -112,10 +117,10 @@ def to_disk(self, filename: pathlib.Path, tags: List[str] = None):
                 json.dump(manifest, outfile)
 
             if isinstance(filename, io.BytesIO):
-                output_kwargs = {"fileobj": filename, "mode": "w"}
+                tar_open = tarfile.open(fileobj=filename, mode="w")
             else:
-                output_kwargs = {"name": filename, "mode": "w"}
-            with tarfile.open(**output_kwargs) as tar_out:
+                tar_open = tarfile.open(name=filename, mode="w")
+            with tar_open as tar_out:
                 os.chdir(temp_dir)
                 tar_out.add(".")
                 os.chdir("..")