From b5faecbdf12cec89ef7a2452e41477f185066da0 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Sun, 12 Apr 2026 22:30:38 +0200 Subject: [PATCH 1/3] refactor: Cover all easy cases of mypy typing. --- .pre-commit-config.yaml | 8 +++ crpy/__init__.py | 4 +- crpy/auth.py | 10 +-- crpy/cmd.py | 1 - crpy/common.py | 21 +++---- crpy/image.py | 51 +++++++++------- crpy/registry.py | 102 +++++++++++++++---------------- crpy/storage.py | 14 ++--- pyproject.toml | 13 ++++ tests/test_registry_endpoints.py | 1 - 10 files changed, 122 insertions(+), 103 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4f7f695..7658eef 100644 --- a/.pre-commit-config.yaml +++ b/.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: diff --git a/crpy/__init__.py b/crpy/__init__.py index 539249d..fc9e941 100644 --- a/crpy/__init__.py +++ b/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__ = [ diff --git a/crpy/auth.py b/crpy/auth.py index 6f5567b..41d0d11 100644 --- a/crpy/auth.py +++ b/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}"} diff --git a/crpy/cmd.py b/crpy/cmd.py index 2f786cf..6990c17 100644 --- a/crpy/cmd.py +++ b/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 ( diff --git a/crpy/common.py b/crpy/common.py index 221fbef..a177100 100644 --- a/crpy/common.py +++ b/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,7 +13,7 @@ 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) @@ -22,11 +21,11 @@ def json(self) -> dict: 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,7 +37,7 @@ 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: @@ -46,7 +45,7 @@ async def _stream(url, headers: dict = None, aiohttp_kwargs: dict = None): 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. diff --git a/crpy/image.py b/crpy/image.py index 056f019..bbabf5f 100644 --- a/crpy/image.py +++ b/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("..") diff --git a/crpy/registry.py b/crpy/registry.py index 1fca24d..ff368ba 100644 --- a/crpy/registry.py +++ b/crpy/registry.py @@ -9,7 +9,7 @@ import tarfile import tempfile from dataclasses import dataclass, field -from typing import List, Optional, Union, Literal +from typing import Any, Literal, Optional, Union from urllib.parse import urlparse from async_lru import alru_cache @@ -53,7 +53,7 @@ class FirewallEntry: role: str url: str - ips: List[str] = field(default_factory=list) + ips: list[str] = field(default_factory=list) redirect: Optional["FirewallEntry"] = None @property @@ -81,13 +81,13 @@ class RegistryInfo: repository: str tag: str https: bool = True - token: Optional[str] = None + token: str | None = None # networking options - proxy: Optional[str] = None + proxy: str | None = None insecure: bool = False # Where the authentication url was resolved - auth_server_url: Optional[str] = None + auth_server_url: str | None = None @property def _headers(self) -> dict: @@ -98,9 +98,9 @@ def _headers(self) -> dict: @property def _aiohttp_kwargs(self) -> dict: - kwargs = {} + kwargs: dict[str, str | bool] = {} if self.proxy: - kwargs["proxy"] = kwargs + kwargs["proxy"] = self.proxy if self.insecure: kwargs["ssl"] = False return kwargs @@ -110,10 +110,10 @@ async def _request_with_auth( url: str, *, method: str = "post", - params: dict = None, - data: Union[dict, bytes, None] = None, - headers: dict = None, - aiohttp_kwargs: dict = None, + params: dict | None = None, + data: dict | bytes | None = None, + headers: dict | None = None, + aiohttp_kwargs: dict | None = None, ) -> Response: if not headers: headers = {} @@ -146,7 +146,7 @@ def v2_url(self): method = "https" if self.https else "http" return f"{method}://{self.registry}/v2" - def manifest_url(self, reference: Optional[str] = None): + def manifest_url(self, reference: str | None = None): return f"{self.v2_url()}/{self.repository}/manifests/{reference or self.tag}" def blobs_url(self): @@ -163,10 +163,10 @@ def __str__(self): async def auth( self, - www_auth: str = None, - username: str = None, - password: str = None, - b64_token: str = None, + www_auth: str | None = None, + username: str | None = None, + password: str | None = None, + b64_token: str | None = None, use_config: bool = True, ): if www_auth is None: @@ -193,7 +193,7 @@ async def auth( return self.token @staticmethod - def from_url(url: str, proxy: str = None, insecure: bool = False) -> "RegistryInfo": + def from_url(url: str, proxy: str | None = None, insecure: bool = False) -> "RegistryInfo": """ Generates a RegistryInfo object from an url, automatically splitting the url into the dataclass fields. @@ -235,7 +235,7 @@ def from_url(url: str, proxy: str = None, insecure: bool = False) -> "RegistryIn return RegistryInfo(registry, name.strip("/"), tag, scheme == "https", proxy=proxy, insecure=insecure) @alru_cache - async def get_manifest(self, fat: bool = False, reference: Optional[str] = None) -> Response: + async def get_manifest(self, fat: bool = False, reference: str | None = None) -> Response: """ Gets the manifest for a remote docker image. This is a JSON file containing the metadata for how the image is stored. @@ -248,7 +248,7 @@ async def get_manifest(self, fat: bool = False, reference: Optional[str] = None) method `get_manifest_from_architecture()` for that. :return: Response object with status code, raw data and response headers. """ - base_headers = ( + base_headers: tuple[str, ...] = ( _schema1_mimetype, _schema2_mimetype, _ociv1_manifest_mimetype, @@ -262,7 +262,7 @@ async def get_manifest(self, fat: bool = False, reference: Optional[str] = None) response = await self._request_with_auth(self.manifest_url(reference), method="get", headers=headers) return response - async def get_manifest_from_architecture(self, architecture: Union[str, Platform, None] = None) -> dict: + async def get_manifest_from_architecture(self, architecture: str | Platform | None = None) -> dict: """ Returns the manifest as a dictionary for a given architecture. If no architecture is specified, it will return the default manifest, which **might contain multiple entries**. If you want to only get the default manifest, @@ -300,7 +300,7 @@ async def get_manifest_from_architecture(self, architecture: Union[str, Platform return manifest.json() @alru_cache - async def get_default_manifest(self, architecture: Union[str, Platform] = None) -> dict: + async def get_default_manifest(self, architecture: str | Platform | None = None) -> dict: """ Same as `get_manifest_from_architecture()`, but will return one and only one manifest. If the architecture is not specified, it will return the default manifest. @@ -316,7 +316,7 @@ async def get_default_manifest(self, architecture: Union[str, Platform] = None) return manifest @alru_cache - async def get_config(self, architecture: Union[str, Platform] = None) -> Response: + async def get_config(self, architecture: str | Platform | None = None) -> Response: """ Gets the config of a docker image. The config contains all basic information of a docker image, including the entrypoints, cmd, environment variables, etc. @@ -333,7 +333,7 @@ async def get_config(self, architecture: Union[str, Platform] = None) -> Respons return response @alru_cache - async def get_layers(self, architecture: Union[str, Platform, None] = None) -> List[str]: + async def get_layers(self, architecture: str | Platform | None = None) -> list[str]: """ Gets the digests for each layer available at the remote registry. :param architecture: optional architecture for the image. If not provided, the default registry architecture @@ -344,9 +344,7 @@ async def get_layers(self, architecture: Union[str, Platform, None] = None) -> L layers = [m["digest"] for m in manifest["layers"]] return layers - async def pull_layer( - self, layer: str, file_obj: Optional[io.BytesIO] = None, use_cache: bool = True - ) -> Optional[bytes]: + async def pull_layer(self, layer: str, file_obj: io.BytesIO | None = None, use_cache: bool = True) -> bytes | None: """ Retrieves a layer from a remote registry. @@ -367,15 +365,13 @@ async def pull_layer( return await self.get_content_from_remote(layer, file_obj, use_cache) - async def get_content_from_remote( - self, layer: str, file_obj: Optional[io.BytesIO], use_cache: bool - ) -> Optional[bytes]: + async def get_content_from_remote(self, layer: str, file_obj: io.BytesIO | None, use_cache: bool) -> bytes | None: content = await self.get_response_content(layer, file_obj) if use_cache: save_layer(layer, content if file_obj is None else file_obj.getvalue()) return content - async def get_response_content(self, layer: str, file_obj: Optional[io.BytesIO]) -> bytes: + async def get_response_content(self, layer: str, file_obj: io.BytesIO | None) -> bytes: if file_obj is None: response = await self._request_with_auth(f"{self.blobs_url()}/{layer}", method="get", headers=self._headers) return response.data @@ -389,8 +385,8 @@ async def get_response_content(self, layer: str, file_obj: Optional[io.BytesIO]) async def pull( self, - output_file: Union[str, pathlib.Path, io.BytesIO], - architecture: Union[str, Platform, None] = None, + output_file: str | pathlib.Path | io.BytesIO, + architecture: str | Platform | None = None, use_cache: bool = True, ): """ @@ -405,20 +401,19 @@ async def pull( :return: """ print(f"{self.tag}: Pulling from {self.registry}/{self.repository}") - image = Image() - image.manifest = await self.get_default_manifest(architecture) + manifest = await self.get_default_manifest(architecture) raw_config = await self.get_config(architecture) - image.config = raw_config.data + config = raw_config.data + layers = [] for layer in await self.get_layers(architecture): layer_without_prefix = layer.split(":")[1] - image.layers.append( - Blob.from_any(await self.pull_layer(layer, use_cache=use_cache), digest=layer_without_prefix) - ) + layers.append(Blob.from_any(await self.pull_layer(layer, use_cache=use_cache), digest=layer_without_prefix)) print(f"{layer_without_prefix[0:12]}: Pull complete") + image = Image(config, manifest, layers) image.to_disk(output_file, tags=[str(self)]) print(f"Downloaded image from {self}") - async def push_layer(self, file_obj: Union[bytes, str, pathlib.Path], force: bool = False) -> Optional[dict]: + async def push_layer(self, file_obj: bytes | str | pathlib.Path, force: bool = False) -> dict: """ Pushes a layer to a remote repo. @@ -458,13 +453,13 @@ async def push_layer(self, file_obj: Union[bytes, str, pathlib.Path], force: boo data=content, headers={"Content-Type": "application/octet-stream"}, ) - assert response.status == 201, f"Failed to upload blob with digest {digest}: {response.data}" + assert response.status == 201, f"Failed to upload blob with digest {digest}: {response.data.decode()}" manifest["existing"] = False return manifest @staticmethod def build_manifest( - config: dict, layers: List[dict], schema_version: int = 2, media_type: str = _schema2_mimetype + config: dict, layers: list[dict], schema_version: int = 2, media_type: str = _schema2_mimetype ) -> dict: """ Manifest generator, taken from: @@ -508,7 +503,7 @@ async def push_manifest(self, manifest: dict) -> Response: assert response.status == 201 return response - async def push(self, input_file: Union[str, pathlib.Path, io.BytesIO]): + async def push(self, input_file: str | pathlib.Path | io.BytesIO): """ Pushes an input file to the remote repository. The tag that will be used is the one defined for the object. If no tag was provided, the default "latest" will be used. The file must be a tar-file with the config, manifest @@ -560,9 +555,9 @@ async def push(self, input_file: Union[str, pathlib.Path, io.BytesIO]): image_digest = r.headers.get("Docker-Content-Digest", "") or r.headers.get("docker-content-digest") print(f"Pushed {self.tag}: digest: {image_digest}") - async def _list(self, path: str, last: str = None, n: int = None, lazy: bool = False) -> List[dict]: + async def _list(self, path: str, last: str | None = None, n: int | None = None, lazy: bool = False) -> list[dict]: url = f"{self.v2_url()}/{path}" - params = {} + params: dict[str, Any] = {} if n is not None: params["n"] = n if last is not None: @@ -571,13 +566,14 @@ async def _list(self, path: str, last: str = None, n: int = None, lazy: bool = F ret_value = [response.json()] # use pagination to get further tags, if any if "Link" in response.headers and not lazy: - last = re.search(r"last=(\w+)", response.headers["Link"]).group(1) - n = re.search(r"n=(\d+)", response.headers["Link"]).group(1) - next_response = await self._list(path, last, int(n)) - ret_value.append(next_response[0]) + last_match = re.search(r"last=(\w+)", response.headers["Link"]) + n_match = re.search(r"n=(\d+)", response.headers["Link"]) + if last_match and n_match: + next_response = await self._list(path, last_match.group(1), int(n_match.group(1))) + ret_value.append(next_response[0]) return ret_value - async def list_repositories(self, last: str = None, n: int = None) -> List[str]: + async def list_repositories(self, last: str | None = None, n: int | None = None) -> list[str]: """ Lists the repositories contents to show all available images. It will retrieve all pages, unless lazy is specified. In order to list, the user token must have the appropriate permissions. @@ -591,7 +587,7 @@ async def list_repositories(self, last: str = None, n: int = None) -> List[str]: response = await self._list("_catalog", last, n, False if n is None else True) return [entry for page in response for entry in page["repositories"]] - async def list_tags(self, last: str = None, n: int = None) -> List[str]: + async def list_tags(self, last: str | None = None, n: int | None = None) -> list[str]: """ Lists the tags available for the repository. It will retrieve all pages, unless lazy is specified. In order to list, the user token must have the appropriate permissions. @@ -636,7 +632,7 @@ async def _resolve_entry(self, role: str, blob_url: str) -> FirewallEntry: async def resolve( self, architecture: Union[str, "Platform", None] = None, ip_version: Literal[4, 6, 0] = 0 - ) -> List[FirewallEntry]: + ) -> list[FirewallEntry]: """ Performs a dry-run pull to discover every network endpoint that a real pull would contact. Executes authentication, manifest fetch, and HEAD requests for config and layer blobs — without downloading any data. @@ -658,7 +654,7 @@ async def resolve( # multiple queries can be done for retrieving the manifest, but they all go to the same url manifest = await self.get_default_manifest(architecture) - entries: List[FirewallEntry] = [] + entries: list[FirewallEntry] = [] if self.auth_server_url: entries.append(FirewallEntry("auth", self.auth_server_url)) entries.append(FirewallEntry("manifest", self.manifest_url())) @@ -672,7 +668,7 @@ async def resolve( entries.append(await self._resolve_entry(f"layer-{idx}", f"{self.blobs_url()}/{layer}")) # unwrap all entries once in case there were any redirects - unwrapped_entries: List[FirewallEntry] = [] + unwrapped_entries: list[FirewallEntry] = [] for entry in entries: unwrapped_entries.append(entry) if entry.redirect: diff --git a/crpy/storage.py b/crpy/storage.py index 2c4c142..12640d6 100644 --- a/crpy/storage.py +++ b/crpy/storage.py @@ -5,7 +5,6 @@ import sys from base64 import b64encode from functools import lru_cache -from typing import Optional, Tuple from rich import print @@ -39,21 +38,22 @@ def get_config() -> dict: return {} -def get_credentials(url: str) -> Optional[str]: +def get_credentials(url: str) -> str | None: creds = get_config() if "auths" in creds and url in creds["auths"] and "auth" in creds["auths"][url]: return creds["auths"][url]["auth"] return None -def decode_credentials(creds: str) -> Tuple[str, str]: +def decode_credentials(creds: str) -> tuple[str, str]: decoded_string = base64.b64decode(creds).decode() - return tuple(decoded_string.split(":", 1)) # noqa + user, password = decoded_string.split(":", 1) + return user, password def save_credentials(url: str, username: str, password: str): creds = get_config() - token = b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + token = b64encode(f"{username}:{password}".encode()).decode("ascii") creds["auths"][url] = {"auth": token} get_config_file().write_text(json.dumps(creds, indent=2)) @@ -65,7 +65,7 @@ def remove_credentials(url: str) -> bool: return removed is not None -def get_layer_path(layer: str) -> Optional[pathlib.Path]: +def get_layer_path(layer: str) -> pathlib.Path | None: cache_dir = get_config_dir() / "blobs/" os.makedirs(cache_dir, exist_ok=True) layer_path = cache_dir / layer.replace(":", "_") @@ -82,7 +82,7 @@ def save_layer(layer: str, layer_data: bytes): file.write(layer_data) -def get_layer_from_cache(layer: str) -> Optional[bytes]: +def get_layer_from_cache(layer: str) -> bytes | None: """Returns the cache in bytes. If missing on disk, returns None.""" layer_path = get_layer_path(layer) if layer_path: diff --git a/pyproject.toml b/pyproject.toml index 8220add..a29fe17 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,3 +55,16 @@ version = { attr = "crpy.version.__version__" } [tool.ruff] line-length = 120 + +[tool.ruff.lint] +select = [ + # pycodestyle: error and warning + "E", + "W", + # pyupgrade + "UP", + # PyFlakes + "F", + # isort + "I", +] diff --git a/tests/test_registry_endpoints.py b/tests/test_registry_endpoints.py index 1f46288..6da7872 100644 --- a/tests/test_registry_endpoints.py +++ b/tests/test_registry_endpoints.py @@ -1,7 +1,6 @@ import io import tarfile - from crpy.common import Platform, compute_sha256 from crpy.registry import RegistryInfo From d8113ac92117dbafa2799f144604b0a8fa8a453a Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Mon, 13 Apr 2026 21:03:04 +0200 Subject: [PATCH 2/3] refactor: Make params type more specific. --- crpy/registry.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crpy/registry.py b/crpy/registry.py index ff368ba..c3b1815 100644 --- a/crpy/registry.py +++ b/crpy/registry.py @@ -9,7 +9,7 @@ import tarfile import tempfile from dataclasses import dataclass, field -from typing import Any, Literal, Optional, Union +from typing import Literal, Optional, Union from urllib.parse import urlparse from async_lru import alru_cache @@ -557,7 +557,7 @@ async def push(self, input_file: str | pathlib.Path | io.BytesIO): async def _list(self, path: str, last: str | None = None, n: int | None = None, lazy: bool = False) -> list[dict]: url = f"{self.v2_url()}/{path}" - params: dict[str, Any] = {} + params: dict[str, int | str] = {} if n is not None: params["n"] = n if last is not None: From 4423135f09af6d2c5fe9f1b9a34560e69809f748 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Mon, 13 Apr 2026 21:11:29 +0200 Subject: [PATCH 3/3] chore: Remove python 3.9 from labels. --- pyproject.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index a29fe17..558beeb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,6 @@ dynamic = ["version"] classifiers = [ "Programming Language :: Python", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12",