diff --git a/mypy.ini b/mypy.ini index 0a9fae1c2..38110b289 100644 --- a/mypy.ini +++ b/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 diff --git a/src/globus_cli/commands/delete.py b/src/globus_cli/commands/delete.py index da35c0fb8..0c8634324 100644 --- a/src/globus_cli/commands/delete.py +++ b/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") diff --git a/src/globus_cli/commands/timer/create/transfer.py b/src/globus_cli/commands/timer/create/transfer.py index 700d59fd7..7958c81d4 100644 --- a/src/globus_cli/commands/timer/create/transfer.py +++ b/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 diff --git a/src/globus_cli/parsing/command_state.py b/src/globus_cli/parsing/command_state.py index 8f34ad7cf..51bf6cdd2 100644 --- a/src/globus_cli/parsing/command_state.py +++ b/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) diff --git a/src/globus_cli/parsing/commands.py b/src/globus_cli/parsing/commands.py index f4918ba91..e0fe06b6c 100644 --- a/src/globus_cli/parsing/commands.py +++ b/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 `.__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 diff --git a/src/globus_cli/parsing/param_types/endpoint_plus_path.py b/src/globus_cli/parsing/param_types/endpoint_plus_path.py index fe62cf7e2..797961620 100644 --- a/src/globus_cli/parsing/param_types/endpoint_plus_path.py +++ b/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. diff --git a/src/globus_cli/parsing/shared_options/__init__.py b/src/globus_cli/parsing/shared_options/__init__.py index d673db8ec..dac64bb40 100644 --- a/src/globus_cli/parsing/shared_options/__init__.py +++ b/src/globus_cli/parsing/shared_options/__init__.py @@ -1,5 +1,6 @@ from __future__ import annotations +import datetime import functools import textwrap import typing as t @@ -20,9 +21,7 @@ C = t.TypeVar("C", bound=t.Union[t.Callable, click.Command]) -def common_options( - f: t.Callable | None = None, *, disable_options: list[str] | None = None -) -> t.Callable: +def common_options(*, disable_options: list[str] | None = None) -> t.Callable[[C], C]: """ This is a multi-purpose decorator for applying a "base" set of options shared by all commands. @@ -44,23 +43,24 @@ def common_options( """ if disable_options is None: disable_options = [] - if f is None: - return functools.partial(common_options, disable_options=disable_options) - f = debug_option(f) - f = show_server_timing_option(f) - f = verbose_option(f) - f = click.help_option("-h", "--help")(f) + def decorator(f: C) -> C: + f = debug_option(f) + f = show_server_timing_option(f) + f = verbose_option(f) + f = click.help_option("-h", "--help")(f) - # if the format option is being allowed, it needs to be applied to `f` - if "format" not in disable_options: - f = format_option(f) + # if the format option is being allowed, it needs to be applied to `f` + if "format" not in disable_options: + f = format_option(f) - # if the --map-http-status option is being allowed, ... - if "map_http_status" not in disable_options: - f = map_http_status_option(f) + # if the --map-http-status option is being allowed, ... + if "map_http_status" not in disable_options: + f = map_http_status_option(f) - return f + return f + + return decorator def task_notify_option(f: C) -> C: @@ -79,12 +79,14 @@ def task_notify_option(f: C) -> C: )(f) -def task_submission_options(f): +def task_submission_options(f: C) -> C: """ Options shared by both transfer and delete task submission """ - def format_deadline_callback(ctx, param, value): + def format_deadline_callback( + ctx: click.Context, param: click.Parameter, value: datetime.datetime | None + ) -> str | None: if not value: return None return value.strftime("%Y-%m-%d %H:%M:%S") @@ -121,70 +123,69 @@ def format_deadline_callback(ctx, param, value): def delete_and_rm_options( - f: t.Callable | None = None, *, supports_batch: bool = True, default_enable_globs: bool = False, -): +) -> t.Callable[[C], C]: """ Options which apply both to `globus delete` and `globus rm` """ - if f is None: - return functools.partial( - delete_and_rm_options, - supports_batch=supports_batch, - default_enable_globs=default_enable_globs, - ) - f = click.option("--recursive", "-r", is_flag=True, help="Recursively delete dirs")( - f - ) - f = click.option( - "--ignore-missing", - "-f", - is_flag=True, - help="Don't throw errors if the file or dir is absent", - )(f) - f = click.option( - "--star-silent", - "--unsafe", - "star_silent", - is_flag=True, - help=( - 'Don\'t prompt when the trailing character is a "*".' - + (" Implicit in --batch" if supports_batch else "") - ), - )(f) - f = click.option( - "--enable-globs/--no-enable-globs", - is_flag=True, - default=default_enable_globs, - show_default=True, - help=( - "Enable expansion of *, ?, and [ ] characters in the last " - "component of file paths, unless they are escaped with " - "a preceding backslash, \\" - ), - )(f) - if supports_batch: - f = click.option( - "--batch", - type=click.File("r"), - help=textwrap.dedent( - """\ - Accept a batch of paths from a file. - Use `-` to read from stdin. - Uses ENDPOINT_ID as passed on the commandline. - - See documentation on "Batch Input" for more information. - """ + def decorator(f: C) -> C: + f = click.option( + "--recursive", "-r", is_flag=True, help="Recursively delete dirs" + )(f) + f = click.option( + "--ignore-missing", + "-f", + is_flag=True, + help="Don't throw errors if the file or dir is absent", + )(f) + f = click.option( + "--star-silent", + "--unsafe", + "star_silent", + is_flag=True, + help=( + 'Don\'t prompt when the trailing character is a "*".' + + (" Implicit in --batch" if supports_batch else "") ), )(f) - return f + f = click.option( + "--enable-globs/--no-enable-globs", + is_flag=True, + default=default_enable_globs, + show_default=True, + help=( + "Enable expansion of *, ?, and [ ] characters in the last " + "component of file paths, unless they are escaped with " + "a preceding backslash, \\" + ), + )(f) + if supports_batch: + f = click.option( + "--batch", + type=click.File("r"), + help=textwrap.dedent( + """\ + Accept a batch of paths from a file. + Use `-` to read from stdin. + + Uses ENDPOINT_ID as passed on the commandline. + + See documentation on "Batch Input" for more information. + """ + ), + )(f) + return f + + return decorator -def synchronous_task_wait_options(f): - def polling_interval_callback(ctx, param, value): +def synchronous_task_wait_options(f: C) -> C: + def polling_interval_callback( + ctx: click.Context, param: click.Parameter, value: int | None + ) -> int | None: if not value: return None @@ -195,7 +196,9 @@ def polling_interval_callback(ctx, param, value): return value - def exit_code_callback(ctx, param, value): + def exit_code_callback( + ctx: click.Context, param: click.Parameter, value: int | None + ) -> int | None: if not value: return None @@ -249,13 +252,13 @@ def exit_code_callback(ctx, param, value): def security_principal_opts( *, - allow_anonymous=False, - allow_all_authenticated=False, - allow_provision=False, -): - def preprocess_security_principals(f): + allow_anonymous: bool = False, + allow_all_authenticated: bool = False, + allow_provision: bool = False, +) -> t.Callable[[C], C]: + def preprocess_security_principals(f: C) -> C: @functools.wraps(f) - def decorator(*args, **kwargs): + def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any: identity = kwargs.pop("identity", None) group = kwargs.pop("group", None) provision_identity = kwargs.pop("provision_identity", None) @@ -290,15 +293,15 @@ def decorator(*args, **kwargs): return f(*args, **kwargs) - return decorator + return t.cast(C, wrapper) - def decorate(f: t.Callable) -> t.Callable: + def decorate(f: C) -> C: # order matters here -- the preprocessor must run after option # application, so it has to be applied first if isinstance(f, click.Command): # if we're decorating a command, put the preprocessor on its # callback, not on `f` itself - f.callback = preprocess_security_principals(f.callback) + f.callback = preprocess_security_principals(t.cast(C, f.callback)) else: # otherwise, we're applying to a function, but other decorators may # have been applied to give it params @@ -352,7 +355,7 @@ def decorate(f: t.Callable) -> t.Callable: return decorate -def no_local_server_option(f): +def no_local_server_option(f: C) -> C: """ Option for commands that start auth flows and might need to disable the default local server behavior diff --git a/src/globus_cli/parsing/shared_options/id_args.py b/src/globus_cli/parsing/shared_options/id_args.py index b42a26737..a105eae4f 100644 --- a/src/globus_cli/parsing/shared_options/id_args.py +++ b/src/globus_cli/parsing/shared_options/id_args.py @@ -1,45 +1,30 @@ """ -These are semi-standard arg names with overridable metavars. +These are standardized arg names. Basic usage: >>> @endpoint_id_arg >>> def command_func(endpoint_id: uuid.UUID): >>> ... - -Override metavar (note that argname is unchanged): - ->>> @endpoint_id_arg(metavar='HOST_ENDPOINT_ID') ->>> def command_func(endpoint_id: uuid.UUID): ->>> ... """ -from __future__ import annotations - -import functools import typing as t import click +C = t.TypeVar("C", bound=t.Union[click.Command, t.Callable[..., t.Any]]) + -def collection_id_arg(f: t.Callable | None = None, *, metavar: str = "COLLECTION_ID"): - if f is None: - return functools.partial(collection_id_arg, metavar=metavar) - return click.argument("collection_id", metavar=metavar, type=click.UUID)(f) +def collection_id_arg(f: C) -> C: + return click.argument("collection_id", metavar="COLLECTION_ID", type=click.UUID)(f) -def endpoint_id_arg(f: t.Callable | None = None, *, metavar: str = "ENDPOINT_ID"): - if f is None: - return functools.partial(endpoint_id_arg, metavar=metavar) - return click.argument("endpoint_id", metavar=metavar, type=click.UUID)(f) +def endpoint_id_arg(f: C) -> C: + return click.argument("endpoint_id", metavar="ENDPOINT_ID", type=click.UUID)(f) -def flow_id_arg(f: t.Callable | None = None, *, metavar: str = "FLOW_ID"): - if f is None: - return functools.partial(flow_id_arg, metavar=metavar) - return click.argument("flow_id", metavar=metavar, type=click.UUID)(f) +def flow_id_arg(f: C) -> C: + return click.argument("flow_id", metavar="FLOW_ID", type=click.UUID)(f) -def run_id_arg(f: t.Callable | None = None, *, metavar: str = "RUN_ID"): - if f is None: - return functools.partial(run_id_arg, metavar=metavar) - return click.argument("run_id", metavar=metavar, type=click.UUID)(f) +def run_id_arg(f: C) -> C: + return click.argument("run_id", metavar="RUN_ID", type=click.UUID)(f) diff --git a/src/globus_cli/parsing/shared_options/transfer_task_options.py b/src/globus_cli/parsing/shared_options/transfer_task_options.py index fa9f2dc92..216480989 100644 --- a/src/globus_cli/parsing/shared_options/transfer_task_options.py +++ b/src/globus_cli/parsing/shared_options/transfer_task_options.py @@ -1,6 +1,5 @@ from __future__ import annotations -import functools import textwrap import typing as t @@ -9,36 +8,23 @@ C = t.TypeVar("C", bound=t.Union[t.Callable, click.Command]) -@t.overload -def sync_level_option(f: C) -> C: - ... +def sync_level_option(*, aliases: tuple[str, ...] = ()) -> t.Callable[[C], C]: + def decorator(f: C) -> C: + return click.option( + "--sync-level", + *aliases, + default=None, + show_default=True, + type=click.Choice( + ("exists", "size", "mtime", "checksum"), case_sensitive=False + ), + help=( + "Specify that only new or modified files should be transferred, " + "depending on which setting is provided." + ), + )(f) - -@t.overload -def sync_level_option(*, aliases: tuple[str, ...]) -> t.Callable[[C], C]: - ... - - -def sync_level_option( - f: C | None = None, *, aliases: tuple[str, ...] = () -) -> t.Callable[[C], C] | C: - if f is None: - return t.cast( - t.Callable[[C], C], functools.partial(sync_level_option, aliases=aliases) - ) - return click.option( - "--sync-level", - *aliases, - default=None, - show_default=True, - type=click.Choice( - ("exists", "size", "mtime", "checksum"), case_sensitive=False - ), - help=( - "Specify that only new or modified files should be transferred, " - "depending on which setting is provided." - ), - )(f) + return decorator def transfer_recursive_option(f: C) -> C: @@ -101,31 +87,17 @@ def skip_source_errors_option(f: C) -> C: )(f) -@t.overload -def preserve_timestamp_option(f: C) -> C: - ... - - -@t.overload -def preserve_timestamp_option(*, aliases: tuple[str, ...]) -> t.Callable[[C], C]: - ... +def preserve_timestamp_option(*, aliases: tuple[str, ...] = ()) -> t.Callable[[C], C]: + def decorator(f: C) -> C: + return click.option( + "--preserve-timestamp", + *aliases, + is_flag=True, + default=False, + help="Preserve file and directory modification times.", + )(f) - -def preserve_timestamp_option( - f: C | None = None, *, aliases: tuple[str, ...] = () -) -> t.Callable[[C], C] | C: - if f is None: - return t.cast( - t.Callable[[C], C], - functools.partial(preserve_timestamp_option, aliases=aliases), - ) - return click.option( - "--preserve-timestamp", - *aliases, - is_flag=True, - default=False, - help="Preserve file and directory modification times.", - )(f) + return decorator def verify_checksum_option(f: C) -> C: @@ -137,28 +109,14 @@ def verify_checksum_option(f: C) -> C: )(f) -@t.overload -def encrypt_data_option(f: C) -> C: - ... - +def encrypt_data_option(*, aliases: tuple[str, ...] = ()) -> t.Callable[[C], C]: + def decorator(f: C) -> C: + return click.option( + "--encrypt-data", + *aliases, + is_flag=True, + default=False, + help="Encrypt data sent through the network.", + )(f) -@t.overload -def encrypt_data_option(*, aliases: tuple[str, ...]) -> t.Callable[[C], C]: - ... - - -def encrypt_data_option( - f: C | None = None, *, aliases: tuple[str, ...] = () -) -> t.Callable[[C], C] | C: - if f is None: - return t.cast( - t.Callable[[C], C], - functools.partial(encrypt_data_option, aliases=aliases), - ) - return click.option( - "--encrypt-data", - *aliases, - is_flag=True, - default=False, - help="Encrypt data sent through the network.", - )(f) + return decorator diff --git a/src/globus_cli/parsing/shell_completion.py b/src/globus_cli/parsing/shell_completion.py index 08f31a109..3f5cbbfa1 100644 --- a/src/globus_cli/parsing/shell_completion.py +++ b/src/globus_cli/parsing/shell_completion.py @@ -1,7 +1,12 @@ +from __future__ import annotations + import os +import typing as t import click +C = t.TypeVar("C", bound=t.Union[t.Callable[..., t.Any], click.Command]) + # pulled by running `_GLOBUS_COMPLETE=source globus` in a bash shell BASH_SHELL_COMPLETER = r""" _globus_completion() { @@ -73,8 +78,8 @@ """ # noqa: E501 -def print_completer_option(f): - def callback(ctx, param, value): +def print_completer_option(f: C) -> C: + def callback(ctx: click.Context, param: click.Parameter, value: str | None) -> None: if not value or ctx.resilient_parsing: return @@ -97,7 +102,7 @@ def callback(ctx, param, value): click.get_current_context().exit(0) - def _compopt(flag, value): + def _compopt(flag: str, value: str) -> t.Callable[[C], C]: return click.option( flag, hidden=True,