typing: Add hints to common commands

Change-Id: I433ffc933fef92a879fac9143402519649788f64
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>

Author: stephenfin

openstackclient/common/availability_zone.py

@@ -13,8 +13,11 @@
 
 """Availability Zone action implementations"""
 
+import argparse
 import copy
 import logging
+from collections.abc import Iterable, Sequence
+from typing import Any
 
 from openstack import exceptions as sdk_exceptions
 from openstack import utils as sdk_utils
@@ -27,8 +30,10 @@
 LOG = logging.getLogger(__name__)
 
 
-def _xform_compute_availability_zone(az, include_extra):
-    result = []
+def _xform_compute_availability_zone(
+    az: Any, include_extra: bool
+) -> list[dict[str, str]]:
+    result: list[dict[str, str]] = []
     zone_info = {
         'zone_name': az.name,
         'zone_status': (
@@ -62,8 +67,8 @@ def _xform_compute_availability_zone(az, include_extra):
     return result
 
 
-def _xform_network_availability_zone(az):
-    result = []
+def _xform_network_availability_zone(az: Any) -> list[dict[str, str]]:
+    result: list[dict[str, str]] = []
     zone_info = {}
     zone_info['zone_name'] = az.name
     zone_info['zone_status'] = az.state
@@ -74,8 +79,8 @@ def _xform_network_availability_zone(az):
     return result
 
 
-def _xform_volume_availability_zone(az):
-    result = []
+def _xform_volume_availability_zone(az: Any) -> list[dict[str, str]]:
+    result: list[dict[str, str]] = []
     zone_info = {
         'zone_name': az.name,
         'zone_status': (
@@ -89,7 +94,7 @@ def _xform_volume_availability_zone(az):
 class ListAvailabilityZone(command.Lister):
     _description = _("List availability zones and their status")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             '--compute',
@@ -117,7 +122,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def _get_compute_availability_zones(self, parsed_args):
+    def _get_compute_availability_zones(
+        self, parsed_args: argparse.Namespace
+    ) -> list[dict[str, str]]:
         compute_client = self.app.client_manager.compute
         try:
             data = list(compute_client.availability_zones(details=True))
@@ -132,7 +139,9 @@ def _get_compute_availability_zones(self, parsed_args):
             result += _xform_compute_availability_zone(zone, parsed_args.long)
         return result
 
-    def _get_network_availability_zones(self, parsed_args):
+    def _get_network_availability_zones(
+        self, parsed_args: argparse.Namespace
+    ) -> list[dict[str, str]]:
         network_client = self.app.client_manager.network
         try:
             # Verify that the extension exists.
@@ -153,7 +162,9 @@ def _get_network_availability_zones(self, parsed_args):
             result += _xform_network_availability_zone(zone)
         return result
 
-    def _get_volume_availability_zones(self, parsed_args):
+    def _get_volume_availability_zones(
+        self, parsed_args: argparse.Namespace
+    ) -> list[dict[str, str]]:
         volume_client = sdk_utils.ensure_service_version(
             self.app.client_manager.sdk_connection.volume, '3'
         )
@@ -174,7 +185,9 @@ def _get_volume_availability_zones(self, parsed_args):
             result += _xform_volume_availability_zone(zone)
         return result
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[Sequence[str], Iterable[tuple[Any, ...]]]:
         columns: tuple[str, ...] = ('Zone Name', 'Zone Status')
         if parsed_args.long:
             columns += (

openstackclient/common/clientmanager.py

@@ -70,10 +70,10 @@ class ClientManager(clientmanager.ClientManager):
 
     def __init__(
         self,
-        cli_options=None,
-        api_version=None,
-        pw_func=None,
-    ):
+        cli_options: Any = None,
+        api_version: Any = None,
+        pw_func: Any = None,
+    ) -> None:
         super().__init__(
             cli_options=cli_options,
             api_version=api_version,
@@ -89,7 +89,7 @@ def __init__(
         # store original auth_type
         self._original_auth_type = cli_options.auth_type
 
-    def setup_auth(self):
+    def setup_auth(self) -> None:
         """Set up authentication"""
 
         if self._auth_setup_completed:
@@ -125,7 +125,7 @@ def setup_auth(self):
 
         return super().setup_auth()
 
-    def _fallback_load_auth_plugin(self, e):
+    def _fallback_load_auth_plugin(self, e: Exception) -> None:
         # NOTES(RuiChen): Hack to avoid auth plugins choking on data they don't
         #                 expect, delete fake token and endpoint, then try to
         #                 load auth plugin again with user specified options.
@@ -150,20 +150,20 @@ def _fallback_load_auth_plugin(self, e):
         else:
             raise e
 
-    def is_network_endpoint_enabled(self):
+    def is_network_endpoint_enabled(self) -> bool:
         """Check if the network endpoint is enabled"""
         # NOTE(dtroyer): is_service_available() can also return None if
         #                there is no Service Catalog, callers here are
         #                not expecting that so fold None into True to
         #                use Network API by default
         return self.is_service_available('network') is not False
 
-    def is_compute_endpoint_enabled(self):
+    def is_compute_endpoint_enabled(self) -> bool:
         """Check if Compute endpoint is enabled"""
         return self.is_service_available('compute') is not False
 
     # TODO(stephenfin): Drop volume_client argument in OSC 8.0 or later.
-    def is_volume_endpoint_enabled(self, volume_client=None):
+    def is_volume_endpoint_enabled(self, volume_client: Any = None) -> bool:
         """Check if volume endpoint is enabled"""
         # We check against the service type and all aliases defined by the
         # Service Types Authority
@@ -195,7 +195,7 @@ class PluginModule(Protocol):
 
 
 def _on_load_failure_callback(
-    manager: stevedore.ExtensionManager,
+    manager: stevedore.ExtensionManager[PluginModule],
     ep: importlib.metadata.EntryPoint,
     err: BaseException,
 ) -> None:
@@ -204,7 +204,7 @@ def _on_load_failure_callback(
     )
 
 
-def get_plugin_modules(group):
+def get_plugin_modules(group: str) -> list[Any]:
     """Find plugin entry points"""
     mod_list = []
     mgr: stevedore.ExtensionManager[PluginModule]
@@ -238,7 +238,9 @@ def get_plugin_modules(group):
     return mod_list
 
 
-def build_plugin_option_parser(parser):
+def build_plugin_option_parser(
+    parser: ArgumentParserT,
+) -> ArgumentParserT:
     """Add plugin options to the parser"""
 
     # Loop through extensions to get parser additions

openstackclient/common/configuration.py

@@ -13,6 +13,10 @@
 
 """Configuration action implementations"""
 
+import argparse
+from collections.abc import Iterable, Sequence
+from typing import Any
+
 from keystoneauth1.loading import base
 
 from openstackclient import command
@@ -26,7 +30,7 @@ class ShowConfiguration(command.ShowOne):
 
     auth_required = False
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         mask_group = parser.add_mutually_exclusive_group()
         mask_group.add_argument(
@@ -44,7 +48,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[Sequence[str], Iterable[Any]]:
         info = self.app.client_manager.get_configuration()
 
         # Assume a default secret list in case we do not have an auth_plugin
@@ -68,4 +74,5 @@ def take_action(self, parsed_args):
                 if secret_opt in info:
                     info[secret_opt] = REDACTED
 
-        return zip(*sorted(info.items()))
+        col_headers, col_data = zip(*sorted(info.items()))
+        return col_headers, col_data

openstackclient/common/envvars.py

@@ -15,7 +15,7 @@
 from openstackclient.i18n import _
 
 
-def bool_from_str(value, strict=False):
+def bool_from_str(value: bool | str, strict: bool = False) -> bool:
     true_strings = ('1', 't', 'true', 'on', 'y', 'yes')
     false_strings = ('0', 'f', 'false', 'off', 'n', 'no')
 
@@ -39,7 +39,7 @@ def bool_from_str(value, strict=False):
     raise ValueError(msg)
 
 
-def boolenv(*vars, default=False):
+def boolenv(*vars: str, default: bool = False) -> bool:
     """Search for the first defined of possibly many bool-like env vars.
 
     Returns the first environment variable defined in vars, or returns the

openstackclient/common/extension.py

@@ -15,7 +15,10 @@
 
 """Extension action implementations"""
 
+import argparse
 import logging
+from collections.abc import Iterable, Sequence
+from typing import Any
 
 from osc_lib import utils
 
@@ -25,7 +28,9 @@
 LOG = logging.getLogger(__name__)
 
 
-def _get_extension_columns(item):
+def _get_extension_columns(
+    item: Any,
+) -> tuple[tuple[str, ...], tuple[str, ...]]:
     column_map = {
         'updated': 'updated_at',
     }
@@ -38,7 +43,7 @@ def _get_extension_columns(item):
 class ListExtension(command.Lister):
     _description = _("List API extensions")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             '--compute',
@@ -72,7 +77,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[Sequence[str], Iterable[tuple[Any, ...]]]:
         columns: tuple[str, ...] = ('Name', 'Alias', 'Description')
         if parsed_args.long:
             columns += ('Namespace', 'Updated At', 'Links')
@@ -139,7 +146,7 @@ def take_action(self, parsed_args):
 class ShowExtension(command.ShowOne):
     _description = _("Show API extension")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'extension',
@@ -152,7 +159,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[Sequence[str], Iterable[Any]]:
         client = self.app.client_manager.network
 
         extension = client.find_extension(

openstackclient/common/limits.py

@@ -15,7 +15,10 @@
 
 """Limits Action Implementation"""
 
+import argparse
 import itertools
+from collections.abc import Iterable, Sequence
+from typing import Any
 
 from osc_lib import utils
 
@@ -24,8 +27,8 @@
 from openstackclient.identity import common as identity_common
 
 
-def _format_absolute_limit(absolute_limits):
-    info = {}
+def _format_absolute_limit(absolute_limits: Any) -> dict[str, Any]:
+    info: dict[str, Any] = {}
 
     for key in set(absolute_limits):
         if key in ('id', 'name', 'location'):
@@ -36,7 +39,7 @@ def _format_absolute_limit(absolute_limits):
     return info
 
 
-def _format_rate_limit(rate_limits):
+def _format_rate_limit(rate_limits: Any) -> Any:
     # flatten this:
     #
     #   {'uri': '<uri>', 'limit': [{'value': '<value>', ...], ...}
@@ -52,7 +55,7 @@ def _format_rate_limit(rate_limits):
 class ShowLimits(command.Lister):
     _description = _("Show compute and block storage limits")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         type_group = parser.add_mutually_exclusive_group(required=True)
         type_group.add_argument(
@@ -100,7 +103,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[Sequence[str], Iterable[tuple[Any, ...]]]:
         project_id = None
         if parsed_args.project is not None:
             identity_client = self.app.client_manager.identity