Tighten mypy: globus_cli.parsing

mypy.ini

@@ -11,33 +11,6 @@ warn_no_return = true
 no_implicit_optional = true
 disallow_untyped_defs = true
 
-[mypy-globus_cli.parsing.command_state]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.commands]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.one_use_option]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.param_types.endpoint_plus_path]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.shared_options]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.shared_options.endpoint_create_and_update]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.shared_options.id_args]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.shared_options.transfer_task_options]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.parsing.shell_completion]
-disallow_untyped_defs = false
-
 [mypy-globus_cli.services.gcs]
 disallow_untyped_defs = false
 

src/globus_cli/commands/delete.py

@@ -71,7 +71,7 @@
 """,
 )
 @task_submission_options
-@delete_and_rm_options
+@delete_and_rm_options()
 @local_user_option
 @click.argument("endpoint_plus_path", type=ENDPOINT_PLUS_OPTPATH)
 @LoginManager.requires_login("transfer")

src/globus_cli/commands/timer/create/transfer.py

@@ -79,11 +79,11 @@ def resolve_optional_local_time(
     "destination", metavar="DEST_ENDPOINT_ID[:DEST_PATH]", type=ENDPOINT_PLUS_OPTPATH
 )
 @transfer_batch_option
-@sync_level_option
+@sync_level_option()
 @transfer_recursive_option
-@encrypt_data_option
+@encrypt_data_option()
 @verify_checksum_option
-@preserve_timestamp_option
+@preserve_timestamp_option()
 @skip_source_errors_option
 @fail_on_quota_errors_option
 @task_notify_option

src/globus_cli/parsing/command_state.py

@@ -16,7 +16,7 @@
 F = t.TypeVar("F", bound=t.Union[t.Callable, click.Command])
 
 
-def _setup_logging(level="DEBUG"):
+def _setup_logging(level: str = "DEBUG") -> None:
     conf = {
         "version": 1,
         "formatters": {
@@ -68,7 +68,7 @@ def is_verbose(self) -> bool:
 
 
 def format_option(f: F) -> F:
-    def callback(ctx, param, value):
+    def callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
         if not value:
             return
 
@@ -80,7 +80,9 @@ def callback(ctx, param, value):
 
         state.output_format = value.lower()
 
-    def jmespath_callback(ctx, param, value):
+    def jmespath_callback(
+        ctx: click.Context, param: click.Parameter, value: t.Any
+    ) -> None:
         if value is None:
             return
 
@@ -116,7 +118,7 @@ def jmespath_callback(ctx, param, value):
 
 
 def debug_option(f: F) -> F:
-    def callback(ctx, param, value):
+    def callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
         if not value or ctx.resilient_parsing:
             # turn off warnings altogether
             _warnings.simplefilter("ignore")
@@ -138,7 +140,7 @@ def callback(ctx, param, value):
 
 
 def verbose_option(f: F) -> F:
-    def callback(ctx, param, value):
+    def callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
         # set state verbosity value from option
         state = ctx.ensure_object(CommandState)
         state.verbosity = value
@@ -186,7 +188,7 @@ def callback(ctx, param, value):
 def map_http_status_option(f: F) -> F:
     exit_stat_set = [0, 1] + list(range(50, 100))
 
-    def per_val_callback(ctx, value):
+    def per_val_callback(ctx: click.Context, value: str | None) -> None:
         if value is None:
             return None
         state = ctx.ensure_object(CommandState)
@@ -197,12 +199,12 @@ def per_val_callback(ctx, value):
             # iterate over those pairs, splitting them on `=` signs
             for http_stat, exit_stat in (pair.split("=") for pair in pairs):
                 # "parse" as ints
-                http_stat, exit_stat = int(http_stat), int(exit_stat)
+                http_stat_int, exit_stat_int = int(http_stat), int(exit_stat)
                 # force into the desired range
-                if exit_stat not in exit_stat_set:
+                if exit_stat_int not in exit_stat_set:
                     raise ValueError()
                 # map the status
-                state.http_status_map[http_stat] = exit_stat
+                state.http_status_map[http_stat_int] = exit_stat_int
         # two conditions can cause ValueError: split didn't give right number
         # of args, or results weren't int()-able
         except ValueError:
@@ -212,7 +214,7 @@ def per_val_callback(ctx, value):
                 "0,1,50-99"
             )
 
-    def callback(ctx, param, value):
+    def callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
         """
         Wrap the per-value callback -- multiple=True means that the value is
         always a tuple of given vals.
@@ -233,7 +235,7 @@ def callback(ctx, param, value):
 
 
 def show_server_timing_option(f: F) -> F:
-    def callback(ctx, param, value):
+    def callback(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
         if not value:
             return
         state = ctx.ensure_object(CommandState)

src/globus_cli/parsing/commands.py

@@ -10,6 +10,7 @@
 import importlib
 import logging
 import sys
+import typing as t
 from shutil import get_terminal_size
 
 import click
@@ -20,6 +21,8 @@
 from .shared_options import common_options
 from .shell_completion import print_completer_option
 
+C = t.TypeVar("C", bound=t.Union[click.Command, t.Callable[..., t.Any]])
+
 log = logging.getLogger(__name__)
 
 
@@ -60,7 +63,7 @@ def example_command(*, foo_bar: list[tuple[Literal["foo", "bar"], Any]]):
     you will need to manually activate the endpoint. See 'globus endpoint activate'
     for more details."""
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
         self.adoc_output = kwargs.pop("adoc_output", None)
         self.adoc_examples = kwargs.pop("adoc_examples", None)
         self.globus_disable_opts = kwargs.pop("globus_disable_opts", [])
@@ -84,9 +87,12 @@ def __init__(self, *args, **kwargs):
                 pass
         super().__init__(*args, **kwargs)
 
-    def invoke(self, ctx):
+    def invoke(self, ctx: click.Context) -> t.Any:
         log.debug("command invoke start")
         try:
+            # this isn't forcing interactive mode, it's just checking to see that
+            # the GLOBUS_CLI_INTERACTIVE value is *valid*
+            env_interactive(raising=True)
             return super().invoke(ctx)
         finally:
             log.debug("command invoke exit")
@@ -125,12 +131,6 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
             raise
 
 
-class GlobusCommandEnvChecks(GlobusCommand):
-    def invoke(self, ctx):
-        env_interactive(raising=True)
-        return super().invoke(ctx)
-
-
 class GlobusCommandGroup(click.Group):
     """
     This is a click.Group with any customizations which we deem necessary
@@ -144,10 +144,10 @@ class GlobusCommandGroup(click.Group):
 
     def __init__(
         self,
-        *args,
+        *args: t.Any,
         lazy_subcommands: dict[str, tuple[str, str]] | None = None,
-        **kwargs,
-    ):
+        **kwargs: t.Any,
+    ) -> None:
         super().__init__(*args, **kwargs)
         # lazy_subcommands is a map of the form:
         #
@@ -186,7 +186,7 @@ def _lazy_load(self, ctx: click.Context, cmd_name: str) -> click.Command:
             )
         return cmd_object
 
-    def invoke(self, ctx):
+    def invoke(self, ctx: click.Context) -> t.Any:
         # if no subcommand was given (but, potentially, flags were passed),
         # ctx.protected_args will be empty
         # improves upon the built-in detection given on click.Group by
@@ -208,53 +208,45 @@ class TopLevelGroup(GlobusCommandGroup):
     passes them to a custom error handler.
     """
 
-    def invoke(self, ctx):
+    def invoke(self, ctx: click.Context) -> t.Any:
         try:
             return super().invoke(ctx)
         except Exception:
-            custom_except_hook(sys.exc_info())
+            # mypy thinks that exc_info could be (None, None, None), but... nope. False.
+            custom_except_hook(sys.exc_info())  # type: ignore[arg-type]
 
 
-def main_group(**kwargs):
-    def decorator(f):
-        f = click.group("globus", cls=TopLevelGroup, **kwargs)(f)
-        f = common_options(f)
-        f = print_completer_option(f)
-        return f
+def main_group(**kwargs: t.Any) -> t.Callable[[C], TopLevelGroup]:
+    def decorator(f: C) -> TopLevelGroup:
+        grp = click.group("globus", cls=TopLevelGroup, **kwargs)(f)
+        grp = common_options()(grp)
+        grp = print_completer_option(grp)
+        return grp
 
     return decorator
 
 
-def command(*args, **kwargs):
+def command(*args: t.Any, **kwargs: t.Any) -> t.Callable[[C], GlobusCommand]:
     """
     A helper for decorating commands a-la `click.command`, but pulling the help string
     from `<function>.__doc__` by default.
 
     Also allows the use of custom arguments, which are stored on the command, as in
     "adoc_examples".
-
-    `skip_env_checks` is used to disable environment variable validation prior to
-    running a command, but is ignored when a specific `cls` argument is passed.
     """
     disable_opts = kwargs.pop("disable_options", [])
 
-    def _inner_decorator(func):
-        if "cls" not in kwargs:
-            if kwargs.get("skip_env_checks", False) is True:
-                kwargs["cls"] = GlobusCommand
-            else:
-                kwargs["cls"] = GlobusCommandEnvChecks
-
+    def _inner_decorator(func: C) -> GlobusCommand:
         kwargs["globus_disable_opts"] = disable_opts
 
         return common_options(disable_options=disable_opts)(
-            click.command(*args, **kwargs)(func)
+            click.command(*args, cls=GlobusCommand, **kwargs)(func)
         )
 
     return _inner_decorator
 
 
-def group(*args, **kwargs):
+def group(*args: t.Any, **kwargs: t.Any) -> t.Callable[[C], GlobusCommandGroup]:
     """
     Wrapper over click.group which sets GlobusCommandGroup as the Class
 
@@ -266,9 +258,9 @@ def group(*args, **kwargs):
     """
     disable_opts = kwargs.pop("disable_options", [])
 
-    def inner_decorator(f):
-        f = click.group(*args, cls=GlobusCommandGroup, **kwargs)(f)
-        f = common_options(disable_options=disable_opts)(f)
-        return f
+    def inner_decorator(f: C) -> GlobusCommandGroup:
+        grp = click.group(*args, cls=GlobusCommandGroup, **kwargs)(f)
+        grp = common_options(disable_options=disable_opts)(grp)
+        return grp
 
     return inner_decorator

src/globus_cli/parsing/param_types/endpoint_plus_path.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+import typing as t
 import uuid
 
 import click
@@ -13,7 +16,7 @@ class EndpointPlusPath(click.ParamType):
 
     name = "endpoint plus path"
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
         # path requirement defaults to True, but can be tweaked by a kwarg
         self.path_required = kwargs.pop("path_required", True)
 
@@ -25,14 +28,14 @@ def get_type_annotation(self, param: click.Parameter) -> type:
         else:
             return tuple[uuid.UUID, str | None]
 
-    def get_metavar(self, param):
+    def get_metavar(self, param: click.Parameter | None) -> str:
         """
         Default metavar for this instance of the type.
         """
         return self.metavar
 
     @property
-    def metavar(self):
+    def metavar(self) -> str:
         """
         Metavar as a property, so that we can make it different if `path_required`
         """
@@ -41,7 +44,12 @@ def metavar(self):
         else:
             return "ENDPOINT_ID[:PATH]"
 
-    def convert(self, value, param, ctx):
+    def convert(
+        self,
+        value: tuple[uuid.UUID, str | None] | str,
+        param: click.Parameter | None,
+        ctx: click.Context | None,
+    ) -> tuple[uuid.UUID, str | None]:
         """
         ParamType.convert() is the actual processing method that takes a
         provided parameter and parses it.