diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8fac6070..c2e5fe4c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,7 +22,7 @@ repos: hooks: - id: doc8 - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.9.6 + rev: v0.11.8 hooks: - id: ruff args: ['--fix', '--unsafe-fixes'] @@ -35,7 +35,7 @@ repos: - flake8-import-order~=0.18.2 exclude: '^(doc|releasenotes|tools)/.*$' - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.13.0 + rev: v1.15.0 hooks: - id: mypy additional_dependencies: diff --git a/osc_lib/api/api.py b/osc_lib/api/api.py index 0d14c5db..624d6a32 100644 --- a/osc_lib/api/api.py +++ b/osc_lib/api/api.py @@ -47,9 +47,9 @@ class BaseAPI: def __init__( self, - session: ty.Optional[ksa_session.Session] = None, - service_type: ty.Optional[str] = None, - endpoint: ty.Optional[str] = None, + session: ksa_session.Session | None = None, + service_type: str | None = None, + endpoint: str | None = None, **kwargs: ty.Any, ) -> None: """Base object that contains some common API objects and methods @@ -83,7 +83,7 @@ class BaseAPI: self.service_type = service_type self.endpoint = self._munge_endpoint(endpoint) - def _munge_endpoint(self, endpoint: ty.Optional[str]) -> ty.Optional[str]: + def _munge_endpoint(self, endpoint: str | None) -> str | None: """Hook to allow subclasses to massage the passed-in endpoint Hook to massage passed-in endpoints from arbitrary sources, @@ -108,7 +108,7 @@ class BaseAPI: self, method: str, url: str, - session: ty.Optional[ksa_session.Session] = None, + session: ksa_session.Session | None = None, **kwargs: ty.Any, ) -> requests.Response: """Perform call into session @@ -159,10 +159,10 @@ class BaseAPI: def create( self, url: str, - session: ty.Optional[ksa_session.Session] = None, - method: ty.Optional[str] = None, + session: ksa_session.Session | None = None, + method: str | None = None, **params: ty.Any, - ) -> ty.Union[requests.Response, ty.Any]: + ) -> requests.Response | ty.Any: """Create a new resource :param string url: @@ -185,7 +185,7 @@ class BaseAPI: def delete( self, url: str, - session: ty.Optional[ksa_session.Session] = None, + session: ksa_session.Session | None = None, **params: ty.Any, ) -> requests.Response: """Delete a resource @@ -201,12 +201,12 @@ class BaseAPI: def list( self, path: str, - session: ty.Optional[ksa_session.Session] = None, + session: ksa_session.Session | None = None, body: ty.Any = None, detailed: bool = False, - headers: ty.Optional[dict[str, str]] = None, + headers: dict[str, str] | None = None, **params: ty.Any, - ) -> ty.Union[requests.Response, ty.Any]: + ) -> requests.Response | ty.Any: """Return a list of resources GET ${ENDPOINT}/${PATH}?${PARAMS} @@ -258,9 +258,9 @@ class BaseAPI: def find_attr( self, path: str, - value: ty.Optional[str] = None, - attr: ty.Optional[str] = None, - resource: ty.Optional[str] = None, + value: str | None = None, + attr: str | None = None, + resource: str | None = None, ) -> ty.Any: """Find a resource via attribute or ID @@ -324,7 +324,7 @@ class BaseAPI: def find_bulk( self, path: str, - headers: ty.Optional[dict[str, str]] = None, + headers: dict[str, str] | None = None, **kwargs: ty.Any, ) -> builtins.list[ty.Any]: """Bulk load and filter locally @@ -376,9 +376,9 @@ class BaseAPI: def find( self, path: str, - value: ty.Optional[str] = None, - attr: ty.Optional[str] = None, - headers: ty.Optional[dict[str, str]] = None, + value: str | None = None, + attr: str | None = None, + headers: dict[str, str] | None = None, ) -> ty.Any: """Find a single resource by name or ID diff --git a/osc_lib/api/auth.py b/osc_lib/api/auth.py index 3bf20a35..bc3c3e70 100644 --- a/osc_lib/api/auth.py +++ b/osc_lib/api/auth.py @@ -223,10 +223,10 @@ def build_auth_plugins_option_parser( def get_keystone2keystone_auth( local_auth: identity_base.BaseIdentityPlugin, service_provider: str, - project_id: ty.Optional[str] = None, - project_name: ty.Optional[str] = None, - project_domain_id: ty.Optional[str] = None, - project_domain_name: ty.Optional[str] = None, + project_id: str | None = None, + project_name: str | None = None, + project_domain_id: str | None = None, + project_domain_name: str | None = None, ) -> k2k.Keystone2Keystone: """Return Keystone 2 Keystone authentication for service provider. diff --git a/osc_lib/api/utils.py b/osc_lib/api/utils.py index 45276c46..78cdf224 100644 --- a/osc_lib/api/utils.py +++ b/osc_lib/api/utils.py @@ -20,11 +20,11 @@ _T = ty.TypeVar('_T', bound=list[ty.Any]) def simple_filter( - data: ty.Optional[_T] = None, - attr: ty.Optional[str] = None, - value: ty.Optional[str] = None, - property_field: ty.Optional[str] = None, -) -> ty.Optional[_T]: + data: _T | None = None, + attr: str | None = None, + value: str | None = None, + property_field: str | None = None, +) -> _T | None: """Filter a list of dicts :param list data: diff --git a/osc_lib/cli/format_columns.py b/osc_lib/cli/format_columns.py index 01e57742..0a063bba 100644 --- a/osc_lib/cli/format_columns.py +++ b/osc_lib/cli/format_columns.py @@ -62,7 +62,7 @@ class ListDictColumn(columns.FormattableColumn[list[dict[str, ty.Any]]]): return [dict(x) for x in self._value or []] -class SizeColumn(columns.FormattableColumn[ty.Union[int, float]]): +class SizeColumn(columns.FormattableColumn[int | float]): """Format column for file size content""" def human_readable(self) -> str: diff --git a/osc_lib/cli/identity.py b/osc_lib/cli/identity.py index 4cc819a5..75a48209 100644 --- a/osc_lib/cli/identity.py +++ b/osc_lib/cli/identity.py @@ -12,7 +12,6 @@ # import argparse -import typing as ty from openstack import connection from openstack import exceptions @@ -49,7 +48,7 @@ def add_project_owner_option_to_parser( def find_project( sdk_connection: connection.Connection, name_or_id: str, - domain_name_or_id: ty.Optional[str] = None, + domain_name_or_id: str | None = None, ) -> project.Project: """Find a project by its name name or ID. diff --git a/osc_lib/cli/parseractions.py b/osc_lib/cli/parseractions.py index dc0df5ba..1fbd6b66 100644 --- a/osc_lib/cli/parseractions.py +++ b/osc_lib/cli/parseractions.py @@ -34,8 +34,8 @@ class KeyValueAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: if not isinstance(values, str): raise TypeError('expected str') @@ -68,8 +68,8 @@ class KeyValueAppendAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: if not isinstance(values, str): raise TypeError('expected str') @@ -110,16 +110,16 @@ class MultiKeyValueAction(argparse.Action): self, option_strings: ty.Sequence[str], dest: str, - nargs: ty.Union[int, str, None] = None, - required_keys: ty.Optional[ty.Sequence[str]] = None, - optional_keys: ty.Optional[ty.Sequence[str]] = None, - const: ty.Optional[_T] = None, - default: ty.Union[_T, str, None] = None, - type: ty.Optional[collections.abc.Callable[[str], _T]] = None, - choices: ty.Optional[collections.abc.Iterable[_T]] = None, + nargs: int | str | None = None, + required_keys: ty.Sequence[str] | None = None, + optional_keys: ty.Sequence[str] | None = None, + const: _T | None = None, + default: _T | str | None = None, + type: collections.abc.Callable[[str], _T] | None = None, + choices: collections.abc.Iterable[_T] | None = None, required: bool = False, - help: ty.Optional[str] = None, - metavar: ty.Union[str, tuple[str, ...], None] = None, + help: str | None = None, + metavar: str | tuple[str, ...] | None = None, ) -> None: """Initialize the action object, and parse customized options @@ -203,8 +203,8 @@ class MultiKeyValueAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: if not isinstance(values, str): raise TypeError('expected str') @@ -249,8 +249,8 @@ class MultiKeyValueCommaAction(MultiKeyValueAction): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: """Overwrite the __call__ function of MultiKeyValueAction @@ -308,8 +308,8 @@ class RangeAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: if not isinstance(values, str): msg = _("Invalid range, non-string value provided") @@ -352,10 +352,10 @@ class NonNegativeAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: - if not isinstance(values, (str, int, float)): + if not isinstance(values, str | int | float): msg = _("%s expected a non-negative integer") raise argparse.ArgumentError(self, msg % str(option_string)) diff --git a/osc_lib/clientmanager.py b/osc_lib/clientmanager.py index 872acf08..d1c93175 100644 --- a/osc_lib/clientmanager.py +++ b/osc_lib/clientmanager.py @@ -59,7 +59,7 @@ class ClientCache: class _PasswordHelper(ty.Protocol): - def __call__(self, prompt: ty.Optional[str] = None) -> str: ... + def __call__(self, prompt: str | None = None) -> str: ... class ClientManager: @@ -75,10 +75,10 @@ class ClientManager: def __init__( self, cli_options: cloud_region.CloudRegion, - api_version: ty.Optional[dict[str, str]], - pw_func: ty.Optional[_PasswordHelper] = None, - app_name: ty.Optional[str] = None, - app_version: ty.Optional[str] = None, + api_version: dict[str, str] | None, + pw_func: _PasswordHelper | None = None, + app_name: str | None = None, + app_version: str | None = None, ) -> None: """Set up a ClientManager @@ -212,7 +212,7 @@ class ClientManager: ) @property - def auth_ref(self) -> ty.Optional[ksa_access.AccessInfo]: + def auth_ref(self) -> ksa_access.AccessInfo | None: """Dereference will trigger an auth if it hasn't already""" if ( not self._auth_required @@ -226,11 +226,11 @@ class ClientManager: self._auth_ref = self.auth.get_auth_ref(self.session) return self._auth_ref - def _override_for(self, service_type: str) -> ty.Optional[str]: + def _override_for(self, service_type: str) -> str | None: key = '{}_endpoint_override'.format(service_type.replace('-', '_')) - return ty.cast(ty.Optional[str], self._cli_options.config.get(key)) + return ty.cast(str | None, self._cli_options.config.get(key)) - def is_service_available(self, service_type: str) -> ty.Optional[bool]: + def is_service_available(self, service_type: str) -> bool | None: """Check if a service type is in the current Service Catalog""" # If there is an override, assume the service is available if self._override_for(service_type): @@ -256,9 +256,9 @@ class ClientManager: def get_endpoint_for_service_type( self, service_type: str, - region_name: ty.Optional[str] = None, + region_name: str | None = None, interface: str = 'public', - ) -> ty.Optional[str]: + ) -> str | None: """Return the endpoint URL for the service type.""" # Overrides take priority unconditionally override = self._override_for(service_type) diff --git a/osc_lib/exceptions.py b/osc_lib/exceptions.py index 4494a92d..063291ef 100644 --- a/osc_lib/exceptions.py +++ b/osc_lib/exceptions.py @@ -15,8 +15,6 @@ """Exception definitions.""" -import typing as ty - class CommandError(Exception): pass @@ -66,9 +64,9 @@ class ClientException(Exception): def __init__( self, - code: ty.Union[int, str], - message: ty.Optional[str] = None, - details: ty.Optional[str] = None, + code: int | str, + message: str | None = None, + details: str | None = None, ): if not isinstance(code, int) and message is None: message = code diff --git a/osc_lib/logs.py b/osc_lib/logs.py index 039469e3..9746144a 100644 --- a/osc_lib/logs.py +++ b/osc_lib/logs.py @@ -96,8 +96,8 @@ class _FileFormatter(logging.Formatter): def __init__( self, - options: ty.Optional[argparse.Namespace] = None, - config: ty.Optional[cloud_config.CloudConfig] = None, + options: argparse.Namespace | None = None, + config: cloud_config.CloudConfig | None = None, **kwargs: ty.Any, ) -> None: context = {} diff --git a/osc_lib/shell.py b/osc_lib/shell.py index a69a8fdf..e400f3ef 100644 --- a/osc_lib/shell.py +++ b/osc_lib/shell.py @@ -48,7 +48,7 @@ DEFAULT_DOMAIN = 'default' DEFAULT_INTERFACE = 'public' -def prompt_for_password(prompt: ty.Optional[str] = None) -> str: +def prompt_for_password(prompt: str | None = None) -> str: """Prompt user for a password Prompt for a password if stdin is a tty. @@ -86,15 +86,14 @@ class OpenStackShell(app.App): def __init__( self, - description: ty.Optional[str] = None, - version: ty.Optional[str] = None, - command_manager: ty.Optional[commandmanager.CommandManager] = None, - stdin: ty.Optional[ty.TextIO] = None, - stdout: ty.Optional[ty.TextIO] = None, - stderr: ty.Optional[ty.TextIO] = None, - interactive_app_factory: ty.Optional[ - type['interactive.InteractiveApp'] - ] = None, + description: str | None = None, + version: str | None = None, + command_manager: commandmanager.CommandManager | None = None, + stdin: ty.TextIO | None = None, + stdout: ty.TextIO | None = None, + stderr: ty.TextIO | None = None, + interactive_app_factory: type['interactive.InteractiveApp'] + | None = None, deferred_help: bool = False, ) -> None: # Patch command.Command to add a default auth_required = True @@ -198,9 +197,9 @@ class OpenStackShell(app.App): def build_option_parser( self, - description: ty.Optional[str], - version: ty.Optional[str], - argparse_kwargs: ty.Optional[dict[str, ty.Any]] = None, + description: str | None, + version: str | None, + argparse_kwargs: dict[str, ty.Any] | None = None, ) -> _argparse.ArgumentParser: parser = super().build_option_parser( description, @@ -532,7 +531,7 @@ class OpenStackShell(app.App): self, cmd: 'command.Command', result: int, - err: ty.Optional[BaseException], + err: BaseException | None, ) -> None: self.log.debug('clean_up %s: %s', cmd.__class__.__name__, err or '') @@ -570,7 +569,7 @@ class OpenStackShell(app.App): tcmd.run(targs) -def main(argv: ty.Optional[list[str]] = None) -> int: +def main(argv: list[str] | None = None) -> int: if argv is None: argv = sys.argv[1:] return OpenStackShell().run(argv) diff --git a/osc_lib/tests/utils/test_tags.py b/osc_lib/tests/utils/test_tags.py index 08328027..2ad87a9d 100644 --- a/osc_lib/tests/utils/test_tags.py +++ b/osc_lib/tests/utils/test_tags.py @@ -14,7 +14,6 @@ import argparse import functools -import sys from unittest import mock from osc_lib.tests import utils as test_utils @@ -216,10 +215,7 @@ class TestTagHelps(test_utils.TestCase): :param exp_enhanced: Expected output with ``enhance_help`` set to ``help_enhancer`` """ - if sys.version_info >= (3, 10): - options_name = 'options' - else: - options_name = 'optional arguments' + options_name = 'options' parser = argparse.ArgumentParser( formatter_class=functools.partial(argparse.HelpFormatter, width=78) ) diff --git a/osc_lib/utils/__init__.py b/osc_lib/utils/__init__.py index e600a52a..a72bc685 100644 --- a/osc_lib/utils/__init__.py +++ b/osc_lib/utils/__init__.py @@ -155,7 +155,7 @@ def calculate_header_and_attrs( return column_headers, attrs -def env(*vars: str, **kwargs: ty.Any) -> ty.Optional[str]: +def env(*vars: str, **kwargs: ty.Any) -> str | None: """Search for the first defined of possibly many env vars Returns the first environment variable defined in vars, or @@ -327,9 +327,7 @@ def find_resource( raise exceptions.CommandError(msg % name_or_id) -def format_dict( - data: dict[str, ty.Any], prefix: ty.Optional[str] = None -) -> str: +def format_dict(data: dict[str, ty.Any], prefix: str | None = None) -> str: """Return a formatted string of key value pairs :param data: a dict @@ -358,8 +356,8 @@ def format_dict( def format_dict_of_list( - data: ty.Optional[dict[str, list[ty.Any]]], separator: str = '; ' -) -> ty.Optional[str]: + data: dict[str, list[ty.Any]] | None, separator: str = '; ' +) -> str | None: """Return a formatted string of key value pair :param data: a dict, key is string, value is a list of string, for example: @@ -385,8 +383,8 @@ def format_dict_of_list( def format_list( - data: ty.Optional[list[ty.Any]], separator: str = ', ' -) -> ty.Optional[str]: + data: list[ty.Any] | None, separator: str = ', ' +) -> str | None: """Return a formatted strings :param data: a list of strings @@ -400,8 +398,8 @@ def format_list( def format_list_of_dicts( - data: ty.Optional[list[dict[str, ty.Any]]], -) -> ty.Optional[str]: + data: list[dict[str, ty.Any]] | None, +) -> str | None: """Return a formatted string of key value pairs for each dict :param data: a list of dicts @@ -413,7 +411,7 @@ def format_list_of_dicts( return '\n'.join(format_dict(i) for i in data) -def format_size(size: ty.Union[int, float, None]) -> str: +def format_size(size: int | float | None) -> str: """Display size of a resource in a human readable format :param size: @@ -444,7 +442,7 @@ def format_size(size: ty.Union[int, float, None]) -> str: def get_client_class( api_name: str, - version: ty.Union[str, int, float], + version: str | int | float, version_map: dict[str, type[_T]], ) -> ty.Any: """Returns the client class for the requested API version @@ -485,10 +483,9 @@ def get_client_class( def get_dict_properties( item: dict[str, _T], fields: collections.abc.Sequence[str], - mixed_case_fields: ty.Optional[collections.abc.Sequence[str]] = None, - formatters: ty.Optional[ - dict[str, type[cliff_columns.FormattableColumn[ty.Any]]] - ] = None, + mixed_case_fields: collections.abc.Sequence[str] | None = None, + formatters: dict[str, type[cliff_columns.FormattableColumn[ty.Any]]] + | None = None, ) -> tuple[ty.Any, ...]: """Return a tuple containing the item properties. @@ -537,10 +534,9 @@ def get_dict_properties( def get_item_properties( item: dict[str, _T], fields: collections.abc.Sequence[str], - mixed_case_fields: ty.Optional[collections.abc.Sequence[str]] = None, - formatters: ty.Optional[ - dict[str, type[cliff_columns.FormattableColumn[ty.Any]]] - ] = None, + mixed_case_fields: collections.abc.Sequence[str] | None = None, + formatters: dict[str, type[cliff_columns.FormattableColumn[ty.Any]]] + | None = None, ) -> tuple[ty.Any, ...]: """Return a tuple containing the item properties. @@ -612,7 +608,7 @@ def get_field(item: _T, field: str) -> ty.Any: def get_password( stdin: ty.TextIO, - prompt: ty.Optional[str] = None, + prompt: str | None = None, confirm: bool = True, ) -> str: message = prompt or "User Password:" @@ -634,7 +630,7 @@ def get_password( raise exceptions.CommandError(msg) -def is_ascii(string: ty.Union[str, bytes]) -> bool: +def is_ascii(string: str | bytes) -> bool: try: if isinstance(string, bytes): string.decode('ascii') @@ -658,7 +654,7 @@ def read_blob_file_contents(blob_file: str) -> str: def sort_items( items: collections.abc.Sequence[_T], sort_str: str, - sort_type: ty.Optional[type[ty.Any]] = None, + sort_type: type[ty.Any] | None = None, ) -> collections.abc.Sequence[_T]: """Sort items based on sort keys and sort directions given by sort_str. @@ -724,7 +720,7 @@ def wait_for_delete( exception_name: collections.abc.Sequence[str] = ['NotFound'], sleep_time: int = 5, timeout: int = 300, - callback: ty.Optional[collections.abc.Callable[[int], None]] = None, + callback: collections.abc.Callable[[int], None] | None = None, ) -> bool: """Wait for resource deletion @@ -782,7 +778,7 @@ def wait_for_status( success_status: collections.abc.Sequence[str] = ['active'], error_status: collections.abc.Sequence[str] = ['error'], sleep_time: int = 5, - callback: ty.Optional[collections.abc.Callable[[int], None]] = None, + callback: collections.abc.Callable[[int], None] | None = None, ) -> bool: """Wait for status change on a resource during a long-running operation @@ -820,7 +816,7 @@ def wait_for_status( def get_osc_show_columns_for_sdk_resource( sdk_resource: resource.Resource, osc_column_map: dict[str, str], - invisible_columns: ty.Optional[collections.abc.Sequence[str]] = None, + invisible_columns: collections.abc.Sequence[str] | None = None, ) -> tuple[tuple[str, ...], tuple[str, ...]]: """Get and filter the display and attribute columns for an SDK resource. diff --git a/osc_lib/utils/columns.py b/osc_lib/utils/columns.py index cd1fe2c3..8a4e2678 100644 --- a/osc_lib/utils/columns.py +++ b/osc_lib/utils/columns.py @@ -78,7 +78,7 @@ def get_column_definitions( def get_columns( item: dict[str, ty.Any], - attr_map: ty.Optional[list[tuple[str, str, str]]] = None, + attr_map: list[tuple[str, str, str]] | None = None, ) -> tuple[tuple[str, ...], tuple[str, ...]]: """Return pair of resource attributes and corresponding display names. diff --git a/osc_lib/utils/tags.py b/osc_lib/utils/tags.py index 9dcc3a1a..6377119e 100644 --- a/osc_lib/utils/tags.py +++ b/osc_lib/utils/tags.py @@ -22,8 +22,8 @@ class _CommaListAction(argparse.Action): self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: ty.Union[str, ty.Sequence[ty.Any], None], - option_string: ty.Optional[str] = None, + values: str | ty.Sequence[ty.Any] | None, + option_string: str | None = None, ) -> None: if not isinstance(values, str): raise TypeError('expected str') diff --git a/pyproject.toml b/pyproject.toml index 21a1e995..4bfce639 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [tool.mypy] -python_version = "3.9" +python_version = "3.10" show_column_numbers = true show_error_context = true ignore_missing_imports = true @@ -33,6 +33,7 @@ ignore_errors = true [tool.ruff] line-length = 79 +target-version = "py310" [tool.ruff.format] quote-style = "preserve"