typing: Add hints to compute commands

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

Author: stephenfin

openstackclient/compute/client.py

@@ -12,7 +12,9 @@
 #   License for the specific language governing permissions and limitations
 #   under the License.
 
+import argparse
 import logging
+from typing import Any
 
 from osc_lib import utils
 
@@ -27,7 +29,7 @@
 API_VERSIONS = ('2', '2.1')
 
 
-def make_client(instance):
+def make_client(instance: Any) -> Any:
     """Returns a compute service client."""
     LOG.debug(
         'Compute client initialized using OpenStack SDK: %s',
@@ -36,7 +38,9 @@ def make_client(instance):
     return instance.sdk_connection.compute
 
 
-def build_option_parser(parser):
+def build_option_parser(
+    parser: argparse.ArgumentParser,
+) -> argparse.ArgumentParser:
     """Hook to add global options"""
     parser.add_argument(
         '--os-compute-api-version',
@@ -48,6 +52,6 @@ def build_option_parser(parser):
     return parser
 
 
-def check_api_version(check_version):
+def check_api_version(check_version: str) -> bool:
     # SDK supports auto-negotiation for us: always return True
     return True

openstackclient/compute/v2/agent.py

@@ -11,11 +11,13 @@
 #   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #   License for the specific language governing permissions and limitations
 #   under the License.
-#
 
 """Agent action implementations"""
 
+import argparse
+from collections.abc import Iterable, Sequence
 import logging
+from typing import Any
 
 from openstack import exceptions as sdk_exceptions
 from osc_lib import exceptions
@@ -36,7 +38,7 @@ class CreateAgent(command.ShowOne):
     23.0.0 (Wallaby) release.
     """
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument("os", metavar="<os>", help=_("Type of OS"))
         parser.add_argument(
@@ -55,7 +57,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]]:
         compute_client = self.app.client_manager.compute
 
         # doing this since openstacksdk has decided not to support this
@@ -76,7 +80,8 @@ def take_action(self, parsed_args):
         sdk_exceptions.raise_from_response(response)
         agent = response.json().get('agent')
 
-        return zip(*sorted(agent.items()))
+        col_headers, col_data = zip(*sorted(agent.items()))
+        return col_headers, col_data
 
 
 class DeleteAgent(command.Command):
@@ -87,14 +92,14 @@ class DeleteAgent(command.Command):
     23.0.0 (Wallaby) release.
     """
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             "id", metavar="<id>", nargs='+', help=_("ID of agent(s) to delete")
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
         result = 0
         for id in parsed_args.id:
@@ -129,7 +134,7 @@ class ListAgent(command.Lister):
     23.0.0 (Wallaby) release.
     """
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             "--hypervisor",
@@ -138,7 +143,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[tuple[str, ...], Iterable[tuple[Any, ...]]]:
         compute_client = self.app.client_manager.compute
         columns = (
             "Agent ID",
@@ -171,7 +178,7 @@ class SetAgent(command.Command):
     23.0.0 (Wallaby) release.
     """
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             "id",
@@ -193,7 +200,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
 
         response = compute_client.get('/os-agents', microversion='2.1')

openstackclient/compute/v2/aggregate.py

@@ -16,6 +16,8 @@
 
 """Compute v2 Aggregate action implementations"""
 
+import argparse
+from collections.abc import Iterable, Sequence
 import logging
 from typing import Any
 
@@ -41,7 +43,9 @@
 }
 
 
-def _get_aggregate_columns(item):
+def _get_aggregate_columns(
+    item: Any,
+) -> tuple[tuple[str, ...], tuple[str, ...]]:
     # To maintain backwards compatibility we need to rename sdk props to
     # whatever OSC was using before
     column_map = {
@@ -56,7 +60,7 @@ def _get_aggregate_columns(item):
 class AddAggregateHost(command.ShowOne):
     _description = _("Add host to aggregate")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -68,7 +72,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]]:
         compute_client = self.app.client_manager.compute
 
         aggregate = compute_client.find_aggregate(
@@ -89,7 +95,7 @@ def take_action(self, parsed_args):
 class CreateAggregate(command.ShowOne):
     _description = _("Create a new aggregate")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             "name", metavar="<name>", help=_("New aggregate name")
@@ -111,7 +117,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]]:
         compute_client = self.app.client_manager.compute
 
         attrs = {'name': parsed_args.name}
@@ -137,7 +145,7 @@ def take_action(self, parsed_args):
 class DeleteAggregate(command.Command):
     _description = _("Delete existing aggregate(s)")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -147,7 +155,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
         result = 0
         for a in parsed_args.aggregate:
@@ -180,7 +188,7 @@ def take_action(self, parsed_args):
 class ListAggregate(command.Lister):
     _description = _("List all aggregates")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             '--long',
@@ -190,7 +198,9 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[tuple[str, ...], Iterable[tuple[Any, ...]]]:
         compute_client = self.app.client_manager.compute
 
         aggregates = list(compute_client.aggregates())
@@ -240,7 +250,7 @@ def take_action(self, parsed_args):
 class RemoveAggregateHost(command.ShowOne):
     _description = _("Remove host from aggregate")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -252,7 +262,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]]:
         compute_client = self.app.client_manager.compute
 
         aggregate = compute_client.find_aggregate(
@@ -273,7 +285,7 @@ def take_action(self, parsed_args):
 class SetAggregate(command.Command):
     _description = _("Set aggregate properties")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -309,7 +321,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
         aggregate = compute_client.find_aggregate(
             parsed_args.aggregate, ignore_missing=False
@@ -345,7 +357,7 @@ def take_action(self, parsed_args):
 class ShowAggregate(command.ShowOne):
     _description = _("Display aggregate details")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -354,7 +366,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]]:
         compute_client = self.app.client_manager.compute
         aggregate = compute_client.find_aggregate(
             parsed_args.aggregate, ignore_missing=False
@@ -374,7 +388,7 @@ def take_action(self, parsed_args):
 class UnsetAggregate(command.Command):
     _description = _("Unset aggregate properties")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             "aggregate",
@@ -394,7 +408,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
         aggregate = compute_client.find_aggregate(
             parsed_args.aggregate, ignore_missing=False
@@ -411,7 +425,7 @@ class CacheImageForAggregate(command.Command):
     # NOTE(gtema): According to stephenfin and dansmith there is no and will
     # not be anything to return.
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'aggregate',
@@ -429,7 +443,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
 
         if not sdk_utils.supports_microversion(compute_client, '2.81'):

openstackclient/compute/v2/console.py

@@ -15,14 +15,18 @@
 
 """Compute v2 Console action implementations"""
 
+import argparse
+from collections.abc import Iterable, Sequence
+from typing import Any
+
 from osc_lib.cli import parseractions
 from osc_lib import utils
 
 from openstackclient import command
 from openstackclient.i18n import _
 
 
-def _get_console_columns(item):
+def _get_console_columns(item: Any) -> tuple[tuple[str, ...], tuple[str, ...]]:
     # To maintain backwards compatibility we need to rename sdk props to
     # whatever OSC was using before
     hidden_columns = ['id', 'links', 'location', 'name']
@@ -34,7 +38,7 @@ def _get_console_columns(item):
 class ShowConsoleLog(command.Command):
     _description = _("Show server's console output")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'server',
@@ -54,7 +58,7 @@ def get_parser(self, prog_name):
         )
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         compute_client = self.app.client_manager.compute
 
         server = compute_client.find_server(
@@ -78,7 +82,7 @@ def take_action(self, parsed_args):
 class ShowConsoleURL(command.ShowOne):
     _description = _("Show server's remote console URL")
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             'server',
@@ -138,7 +142,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]]:
         compute_client = self.app.client_manager.compute
         server = compute_client.find_server(
             parsed_args.server, ignore_missing=False