From 379683962fc59032a3df8421dfea2ab7b4f40a2a Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Mon, 11 May 2026 15:34:14 +0100 Subject: [PATCH] chore: bump pyupgrade to v3.21.2 and apply --py310-plus Regenerates files under the new pyupgrade ruleset. No behavior change. --- .github/.platforms/generate_platforms.py | 3 +- .pre-commit-config.yaml | 10 ++-- cloudsmith_cli/cli/commands/mcp.py | 5 +- cloudsmith_cli/cli/saml.py | 8 ++-- cloudsmith_cli/core/api/init.py | 4 +- cloudsmith_cli/core/api/metadata.py | 12 ++--- cloudsmith_cli/core/download.py | 61 ++++++++++++------------ cloudsmith_cli/core/mcp/data.py | 6 +-- cloudsmith_cli/core/mcp/server.py | 44 ++++++++--------- cloudsmith_cli/core/pagination.py | 11 +++-- 10 files changed, 82 insertions(+), 82 deletions(-) diff --git a/.github/.platforms/generate_platforms.py b/.github/.platforms/generate_platforms.py index 98034044..fdcc05d5 100755 --- a/.github/.platforms/generate_platforms.py +++ b/.github/.platforms/generate_platforms.py @@ -9,9 +9,10 @@ import sys import tempfile import time +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path -from typing import Callable, TypeVar +from typing import TypeVar # Configuration PYTHON_VERSIONS = ("3.10", "3.11", "3.12", "3.13", "3.14") diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9821c3fb..26d261af 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -58,10 +58,12 @@ repos: hooks: - id: isort -- repo: https://github.com/asottile/pyupgrade - rev: v3.19.1 - hooks: - - id: pyupgrade +- repo: https://github.com/asottile/pyupgrade + rev: v3.21.2 + hooks: + - id: pyupgrade + args: [--py310-plus] + - repo: https://github.com/zizmorcore/zizmor-pre-commit rev: v1.20.0 diff --git a/cloudsmith_cli/cli/commands/mcp.py b/cloudsmith_cli/cli/commands/mcp.py index 42053ff9..1cfd7e2b 100644 --- a/cloudsmith_cli/cli/commands/mcp.py +++ b/cloudsmith_cli/cli/commands/mcp.py @@ -5,7 +5,6 @@ import shutil import sys from pathlib import Path -from typing import Dict, List import click import json5 @@ -110,7 +109,7 @@ def list_groups(ctx, opts, mcp_server: server.DynamicMCPServer): print_groups(groups) -def print_tools(tool_list: Dict[str, OpenAPITool]): +def print_tools(tool_list: dict[str, OpenAPITool]): """Print tools as a table or output in another format.""" headers = [ @@ -138,7 +137,7 @@ def print_tools(tool_list: Dict[str, OpenAPITool]): utils.pretty_print_list_info(num_results=num_results, suffix=list_suffix) -def print_groups(group_list: Dict[str, List[str]]): +def print_groups(group_list: dict[str, list[str]]): """Print tool groups as a table or output in another format.""" headers = [ diff --git a/cloudsmith_cli/cli/saml.py b/cloudsmith_cli/cli/saml.py index 450798ed..087b0d7e 100644 --- a/cloudsmith_cli/cli/saml.py +++ b/cloudsmith_cli/cli/saml.py @@ -49,7 +49,7 @@ def get_idp_url(api_host, owner, session): def exchange_2fa_token(api_host, two_factor_token, totp_token, session): exchange_data = {"two_factor_token": two_factor_token, "totp_token": totp_token} - exchange_url = "{api_host}/user/two-factor/".format(api_host=api_host) + exchange_url = f"{api_host}/user/two-factor/" headers = { "Authorization": "Bearer {two_factor_token}".format( @@ -82,11 +82,9 @@ def exchange_2fa_token(api_host, two_factor_token, totp_token, session): def refresh_access_token(api_host, access_token, refresh_token, session): data = {"refresh_token": refresh_token} - url = "{api_host}/user/refresh-token/".format(api_host=api_host) + url = f"{api_host}/user/refresh-token/" - headers = { - "Authorization": "Bearer {access_token}".format(access_token=access_token) - } + headers = {"Authorization": f"Bearer {access_token}"} response = session.post( url, diff --git a/cloudsmith_cli/core/api/init.py b/cloudsmith_cli/core/api/init.py index b6a856bc..407e29e4 100644 --- a/cloudsmith_cli/core/api/init.py +++ b/cloudsmith_cli/core/api/init.py @@ -1,7 +1,7 @@ """Cloudsmith API - Initialisation.""" import base64 -from typing import Type, TypeVar +from typing import TypeVar import click import cloudsmith_api @@ -128,7 +128,7 @@ def initialise_api( T = TypeVar("T") -def get_api_client(cls: Type[T]) -> T: +def get_api_client(cls: type[T]) -> T: """Get an API client (with configuration).""" config = cloudsmith_api.Configuration() client = cls() diff --git a/cloudsmith_cli/core/api/metadata.py b/cloudsmith_cli/core/api/metadata.py index 5091d6f1..f3df8a60 100644 --- a/cloudsmith_cli/core/api/metadata.py +++ b/cloudsmith_cli/core/api/metadata.py @@ -1,7 +1,7 @@ """API - Package metadata (v2) endpoints.""" import json -from typing import Any, Optional, Union +from typing import Any import cloudsmith_api @@ -136,10 +136,10 @@ def _response_json(response): def list_metadata( package_slug_perm: str, *, - source_kind: Optional[Union[int, str]] = None, - classification: Optional[Union[int, str]] = None, - page: Optional[int] = None, - page_size: Optional[int] = None, + source_kind: int | str | None = None, + classification: int | str | None = None, + page: int | None = None, + page_size: int | None = None, ): """List metadata entries attached to a package. @@ -216,7 +216,7 @@ def update_metadata( metadata_slug_perm: str, *, content: Any = None, - source_identity: Optional[str] = None, + source_identity: str | None = None, ): """Patch an existing customer-owned metadata entry. diff --git a/cloudsmith_cli/core/download.py b/cloudsmith_cli/core/download.py index b3638017..578131a3 100644 --- a/cloudsmith_cli/core/download.py +++ b/cloudsmith_cli/core/download.py @@ -3,7 +3,6 @@ import fnmatch import hashlib import os -from typing import Dict, List, Optional, Tuple import click import cloudsmith_api @@ -18,8 +17,8 @@ def resolve_auth( - opts, api_key_opt: Optional[str] = None -) -> Tuple[requests.Session, Dict[str, str], str]: + opts, api_key_opt: str | None = None +) -> tuple[requests.Session, dict[str, str], str]: """ Resolve authentication method and create session with appropriate headers. @@ -56,7 +55,7 @@ def resolve_auth( return session, headers, auth_source -def _matches_tag_filter(pkg: Dict, tag_filter: str) -> bool: +def _matches_tag_filter(pkg: dict, tag_filter: str) -> bool: """ Check if a package matches the tag filter. @@ -84,13 +83,13 @@ def _search_packages( repo: str, name: str, *, - version: Optional[str] = None, - format_filter: Optional[str] = None, - os_filter: Optional[str] = None, - arch_filter: Optional[str] = None, - tag_filter: Optional[str] = None, - filename_filter: Optional[str] = None, -) -> List[Dict]: + version: str | None = None, + format_filter: str | None = None, + os_filter: str | None = None, + arch_filter: str | None = None, + tag_filter: str | None = None, + filename_filter: str | None = None, +) -> list[dict]: """ Search for packages matching criteria, returning all matches. @@ -172,13 +171,13 @@ def resolve_all_packages( repo: str, name: str, *, - version: Optional[str] = None, - format_filter: Optional[str] = None, - os_filter: Optional[str] = None, - arch_filter: Optional[str] = None, - tag_filter: Optional[str] = None, - filename_filter: Optional[str] = None, -) -> List[Dict]: + version: str | None = None, + format_filter: str | None = None, + os_filter: str | None = None, + arch_filter: str | None = None, + tag_filter: str | None = None, + filename_filter: str | None = None, +) -> list[dict]: """ Find all packages matching the criteria. @@ -224,14 +223,14 @@ def resolve_package( repo: str, name: str, *, - version: Optional[str] = None, - format_filter: Optional[str] = None, - os_filter: Optional[str] = None, - arch_filter: Optional[str] = None, - tag_filter: Optional[str] = None, - filename_filter: Optional[str] = None, + version: str | None = None, + format_filter: str | None = None, + os_filter: str | None = None, + arch_filter: str | None = None, + tag_filter: str | None = None, + filename_filter: str | None = None, yes: bool = False, -) -> Dict: +) -> dict: """ Find a single package matching the criteria, handling multiple matches. @@ -295,7 +294,7 @@ def resolve_package( return best_package -def _display_multiple_packages(packages: List[Dict]) -> None: +def _display_multiple_packages(packages: list[dict]) -> None: """Display a table of multiple matching packages.""" click.echo("Multiple packages found:") click.echo() @@ -319,7 +318,7 @@ def _display_multiple_packages(packages: List[Dict]) -> None: click.echo() -def get_download_url(package: Dict) -> str: +def get_download_url(package: dict) -> str: """ Get the download URL for a package. @@ -346,7 +345,7 @@ def get_download_url(package: Dict) -> str: return download_url -def get_package_files(package: Dict) -> List[Dict]: +def get_package_files(package: dict) -> list[dict]: """ Get all downloadable files associated with a package. @@ -389,7 +388,7 @@ def get_package_files(package: Dict) -> List[Dict]: return downloadable_files -def get_package_detail(owner: str, repo: str, identifier: str) -> Dict: +def get_package_detail(owner: str, repo: str, identifier: str) -> dict: """ Get detailed package information including download URLs. @@ -417,7 +416,7 @@ def stream_download( # noqa: C901 outfile: str, session: requests.Session, *, - headers: Optional[Dict[str, str]] = None, + headers: dict[str, str] | None = None, overwrite: bool = False, quiet: bool = False, ) -> None: @@ -522,7 +521,7 @@ def stream_download( # noqa: C901 click.secho("⚠ Checksum mismatch", fg="yellow", err=True) -def _select_best_package(packages: List[Dict]) -> Dict: +def _select_best_package(packages: list[dict]) -> dict: """Select the best package from multiple matches.""" # Sort by version (desc) then by upload date (desc) diff --git a/cloudsmith_cli/core/mcp/data.py b/cloudsmith_cli/core/mcp/data.py index 459c1db1..99e1c43c 100644 --- a/cloudsmith_cli/core/mcp/data.py +++ b/cloudsmith_cli/core/mcp/data.py @@ -1,5 +1,5 @@ from dataclasses import dataclass -from typing import Any, Dict, Optional +from typing import Any @dataclass @@ -10,8 +10,8 @@ class OpenAPITool: description: str method: str path: str - parameters: Dict[str, Any] + parameters: dict[str, Any] base_url: str - query_filter: Optional[str] + query_filter: str | None is_destructive: bool = False is_read_only: bool = False diff --git a/cloudsmith_cli/core/mcp/server.py b/cloudsmith_cli/core/mcp/server.py index 88069fcb..1ff7ba6b 100644 --- a/cloudsmith_cli/core/mcp/server.py +++ b/cloudsmith_cli/core/mcp/server.py @@ -2,7 +2,7 @@ import copy import inspect import json -from typing import Any, Dict, List, Optional +from typing import Any, Optional from urllib import parse import cloudsmith_api @@ -97,7 +97,7 @@ async def list_tools(self) -> list[types.Tool]: return cleaned_tools - def _clean_schema(self, schema: Dict[str, Any]) -> Dict[str, Any]: + def _clean_schema(self, schema: dict[str, Any]) -> dict[str, Any]: """Clean up schema by removing anyOf patterns and other complexities""" if not isinstance(schema, dict): return schema @@ -113,7 +113,7 @@ def _clean_schema(self, schema: Dict[str, Any]) -> Dict[str, Any]: return cleaned - def _clean_property_schema(self, prop_schema: Dict[str, Any]) -> Dict[str, Any]: + def _clean_property_schema(self, prop_schema: dict[str, Any]) -> dict[str, Any]: """Clean individual property schema""" if not isinstance(prop_schema, dict): return prop_schema @@ -180,8 +180,8 @@ def __init__( use_toon=True, allow_destructive_tools=False, debug_mode=False, - allowed_tool_groups: Optional[List[str]] = None, - allowed_tools: Optional[List[str]] = None, + allowed_tool_groups: list[str] | None = None, + allowed_tools: list[str] | None = None, force_all_tools: bool = False, ): mcp_kwargs = {"log_level": "ERROR"} @@ -195,7 +195,7 @@ def __init__( self.allowed_tool_groups = set(allowed_tool_groups or []) self.allowed_tools = set(allowed_tools or []) self.force_all_tools = force_all_tools - self.tools: Dict[str, OpenAPITool] = {} + self.tools: dict[str, OpenAPITool] = {} self.spec = {} async def load_openapi_spec(self): @@ -214,7 +214,7 @@ async def load_openapi_spec(self): self.spec = response.json() await self._generate_tools_from_spec() - def _get_tool_groups(self, tool_name: str) -> List[str]: + def _get_tool_groups(self, tool_name: str) -> list[str]: """ Extract all hierarchical group names from a tool name, excluding action suffixes. @@ -411,7 +411,7 @@ def _get_additional_headers(self): return headers def _get_request_params( - self, url: str, tool: OpenAPITool, arguments: Dict[str, Any] + self, url: str, tool: OpenAPITool, arguments: dict[str, Any] ): """Get params to use for HTTP request based on tool arguments""" @@ -459,7 +459,7 @@ def _get_request_params( return url, query_params, body_params async def _execute_api_call( - self, tool: OpenAPITool, arguments: Dict[str, Any] + self, tool: OpenAPITool, arguments: dict[str, Any] ) -> str: """Execute an API call based on tool definition""" @@ -524,8 +524,8 @@ async def _execute_api_call( await http_client.aclose() def _extract_parameters_from_schema( - self, schema: Dict[str, Any], param_in: str = "body" - ) -> Dict[str, Any]: + self, schema: dict[str, Any], param_in: str = "body" + ) -> dict[str, Any]: """Extract individual parameters from a resolved schema object""" parameters = {} @@ -549,8 +549,8 @@ def _extract_parameters_from_schema( return parameters def _extract_request_body_parameters( - self, request_body: Dict[str, Any] - ) -> Dict[str, Any]: + self, request_body: dict[str, Any] + ) -> dict[str, Any]: """Extract parameters from OpenAPI 3.0 request body with $ref resolution""" parameters = {} @@ -574,7 +574,7 @@ def _extract_request_body_parameters( return parameters - def _extract_body_parameter(self, body_param: Dict[str, Any]) -> Dict[str, str]: + def _extract_body_parameter(self, body_param: dict[str, Any]) -> dict[str, str]: """Extract parameters from Swagger 2.0 body parameter with $ref resolution""" if "schema" not in body_param: @@ -589,10 +589,10 @@ def _create_tool_from_operation( self, method: str, path: str, - operation: Dict[str, Any], + operation: dict[str, Any], path_parameters: list, base_url: str, - ) -> Optional[OpenAPITool]: + ) -> OpenAPITool | None: """Create a tool definition from an OpenAPI operation""" # Generate operation ID @@ -677,7 +677,7 @@ def _create_tool_from_operation( ) def _format_enum_description( - self, enum_values: List[str], original_description: str + self, enum_values: list[str], original_description: str ) -> str: """Format enum values for better tool descriptions""" @@ -691,7 +691,7 @@ def _format_enum_description( return f"Allowed values:\n{enum_list}" - def _resolve_schema_ref(self, ref_string: str) -> Dict[str, Any]: + def _resolve_schema_ref(self, ref_string: str) -> dict[str, Any]: """ Resolve a $ref reference to its actual schema definition @@ -721,7 +721,7 @@ def _resolve_schema_ref(self, ref_string: str) -> Dict[str, Any]: return current - def _resolve_schema(self, schema: Dict[str, Any]) -> Dict[str, Any]: + def _resolve_schema(self, schema: dict[str, Any]) -> dict[str, Any]: """ Recursively resolve a schema, handling $ref references """ @@ -763,17 +763,17 @@ def run(self): except asyncio.CancelledError: print("Server shutdown requested") - def list_tools(self) -> Dict[str, OpenAPITool]: + def list_tools(self) -> dict[str, OpenAPITool]: """Initialize and return list of tools. Useful for debugging""" asyncio.run(self.load_openapi_spec()) return self.tools - def list_groups(self) -> Dict[str, List[str]]: + def list_groups(self) -> dict[str, list[str]]: """Initialize and return list of tool groups with their tools. Useful for debugging""" asyncio.run(self.load_openapi_spec()) # Build a mapping of group -> list of tools - groups: Dict[str, List[str]] = {} + groups: dict[str, list[str]] = {} for tool_name in self.tools: tool_groups = self._get_tool_groups(tool_name) diff --git a/cloudsmith_cli/core/pagination.py b/cloudsmith_cli/core/pagination.py index e7211879..73131ec8 100644 --- a/cloudsmith_cli/core/pagination.py +++ b/cloudsmith_cli/core/pagination.py @@ -1,6 +1,7 @@ """Core pagination utilities.""" -from typing import Any, Callable, List, Optional, Sequence, Tuple +from collections.abc import Callable, Sequence +from typing import Any MAX_PAGE_SIZE = 1000 @@ -80,12 +81,12 @@ def from_headers(cls, headers): def paginate_results( - api_function: Callable[..., Tuple[Sequence[Any], PageInfo]], + api_function: Callable[..., tuple[Sequence[Any], PageInfo]], page_all: bool, page: int, page_size: int = MAX_PAGE_SIZE, **kwargs: Any, -) -> Tuple[List[Any], PageInfo]: +) -> tuple[list[Any], PageInfo]: """Retrieve paginated results. Behaviour: @@ -108,9 +109,9 @@ def paginate_results( # rather than raising. Downstream pretty printers handle an invalid page_info gracefully. return list(results), page_info - all_results: List[Any] = [] + all_results: list[Any] = [] current_page = 1 - last_page_info: Optional[PageInfo] = None + last_page_info: PageInfo | None = None while True: page_results, last_page_info = api_function( page=current_page, page_size=MAX_PAGE_SIZE, **kwargs