Update mypy and strengthen typing for exc hooks

In order to have stronger and clearer type checking enforcement on the
exception hook registry, it has been refactored. Instead of all hooks
being declared in terms of an exception class and/or a condition
function, the hook registry stores all hooks with relevant condition
functions. Exception classes are represented (internally) by building
condition functions as necessary.

Therefore, the hook registry is now an ordered list of the form:

    [(condition, hook)]

which is evaluated via a core loop:

    for condition, hook in registry:
        if condition(error):
            return hook

This is effectively what the logic was before, but it is now declared
and stated as such, in a more readable manner.

One of the improved inferences was achieved by declaring a separate
interface method for registering an SDK-error-class hook. This
indicates to mypy that any condition function will be passed an SDK
error class, which allows mypy to know what attributes will
(minimally) be present.

---

A prereq for some of these changes to work was a mypy update (to v1.7.1).
This resulted in one minor, unrelated change.

mypy.ini

@@ -11,12 +11,6 @@ warn_no_return = true
 no_implicit_optional = true
 disallow_untyped_defs = true
 
-[mypy-globus_cli.exception_handling.hooks]
-disallow_untyped_defs = false
-
-[mypy-globus_cli.exception_handling.registry]
-disallow_untyped_defs = false
-
 [mypy-globus_cli.parsing.command_state]
 disallow_untyped_defs = false
 

src/globus_cli/exception_handling/hooks.py

@@ -11,10 +11,10 @@
 from globus_cli.termio import PrintableErrorField, write_error_info
 from globus_cli.utils import CLIAuthRequirementsError
 
-from .registry import error_handler
+from .registry import error_handler, sdk_error_handler
 
 
-def _pretty_json(data: dict, compact=False) -> str:
+def _pretty_json(data: dict, compact: bool = False) -> str:
     if compact:
         return json.dumps(data, separators=(",", ":"), sort_keys=True)
     return json.dumps(data, indent=2, separators=(",", ": "), sort_keys=True)
@@ -69,9 +69,8 @@ def handle_internal_auth_requirements(
     return None
 
 
-@error_handler(
-    error_class="GlobusAPIError",
-    condition=lambda err: err.info.authorization_parameters,
+@sdk_error_handler(
+    condition=lambda err: bool(err.info.authorization_parameters),
     exit_status=4,
 )
 def session_hook(exception: globus_sdk.GlobusAPIError) -> None:
@@ -92,9 +91,8 @@ def session_hook(exception: globus_sdk.GlobusAPIError) -> None:
     )
 
 
-@error_handler(
-    error_class="GlobusAPIError",
-    condition=lambda err: err.info.consent_required,
+@sdk_error_handler(
+    condition=lambda err: bool(err.info.consent_required),
     exit_status=4,
 )
 def consent_required_hook(exception: globus_sdk.GlobusAPIError) -> int | None:
@@ -174,7 +172,7 @@ def _concrete_consent_required_hook(
     )
 
 
-@error_handler(
+@sdk_error_handler(
     condition=lambda err: (
         (
             isinstance(err, globus_sdk.TransferAPIError)
@@ -200,7 +198,7 @@ def authentication_hook(
     )
 
 
-@error_handler(error_class="TransferAPIError")
+@sdk_error_handler(error_class="TransferAPIError")
 def transferapi_hook(exception: globus_sdk.TransferAPIError) -> None:
     write_error_info(
         "Transfer API Error",
@@ -213,7 +211,7 @@ def transferapi_hook(exception: globus_sdk.TransferAPIError) -> None:
     )
 
 
-@error_handler(
+@sdk_error_handler(
     error_class="SearchAPIError",
     condition=lambda err: err.code == "BadRequest.ValidationError",
 )
@@ -244,7 +242,7 @@ def searchapi_validationerror_hook(exception: globus_sdk.SearchAPIError) -> None
     write_error_info("Search API Error", fields)
 
 
-@error_handler(error_class="SearchAPIError")
+@sdk_error_handler(error_class="SearchAPIError")
 def searchapi_hook(exception: globus_sdk.SearchAPIError) -> None:
     fields = [
         PrintableErrorField("HTTP status", exception.http_status),
@@ -264,7 +262,7 @@ def searchapi_hook(exception: globus_sdk.SearchAPIError) -> None:
     write_error_info("Search API Error", fields)
 
 
-@error_handler(
+@sdk_error_handler(
     error_class="AuthAPIError",
     condition=lambda err: err.message == "invalid_grant",
 )
@@ -283,7 +281,7 @@ def invalidrefresh_hook(exception: globus_sdk.AuthAPIError) -> None:
     )
 
 
-@error_handler(error_class="AuthAPIError")
+@sdk_error_handler(error_class="AuthAPIError")
 def authapi_hook(exception: globus_sdk.AuthAPIError) -> None:
     write_error_info(
         "Auth API Error",
@@ -295,7 +293,7 @@ def authapi_hook(exception: globus_sdk.AuthAPIError) -> None:
     )
 
 
-@error_handler(error_class="GlobusAPIError")  # catch-all
+@sdk_error_handler()  # catch-all
 def globusapi_hook(exception: globus_sdk.GlobusAPIError) -> None:
     write_error_info(
         "Globus API Error",
@@ -307,7 +305,7 @@ def globusapi_hook(exception: globus_sdk.GlobusAPIError) -> None:
     )
 
 
-@error_handler(error_class="GlobusError")
+@sdk_error_handler(error_class="GlobusError")
 def globus_error_hook(exception: globus_sdk.GlobusError) -> None:
     write_error_info(
         "Globus Error",

src/globus_cli/exception_handling/registry.py

@@ -13,26 +13,55 @@
 HOOK_TYPE = t.Callable[[E], t.NoReturn]
 # something which can be decorated to become a hook
 _HOOK_SRC_TYPE = t.Union[t.Callable[[E], None], t.Callable[[E], t.Optional[int]]]
+
 CONDITION_TYPE = t.Callable[[E], bool]
 
-# must cast the registry to avoid type errors around t.List[<nothing>]
-_HOOKLIST_TYPE = t.List[
-    t.Tuple[HOOK_TYPE, t.Union[str, t.Type[Exception]], CONDITION_TYPE]
-]
-_REGISTERED_HOOKS: _HOOKLIST_TYPE = t.cast(_HOOKLIST_TYPE, [])
+_REGISTERED_HOOKS: list[tuple[HOOK_TYPE, CONDITION_TYPE]] = []
+
+
+def sdk_error_handler(
+    *,
+    error_class: str = "GlobusAPIError",
+    condition: t.Callable[[globus_sdk.GlobusAPIError], bool] | None = None,
+    exit_status: int = 1,
+) -> t.Callable[[_HOOK_SRC_TYPE], HOOK_TYPE]:
+    return _error_handler(
+        condition=_build_condition(condition, error_class), exit_status=exit_status
+    )
 
 
 def error_handler(
-    *, error_class=None, condition=None, exit_status: int = 1
+    *,
+    error_class: type[Exception] | None = None,
+    condition: t.Callable[[globus_sdk.GlobusAPIError], bool] | None = None,
+    exit_status: int = 1,
+) -> t.Callable[[_HOOK_SRC_TYPE], HOOK_TYPE]:
+    return _error_handler(
+        condition=_build_condition(condition, error_class), exit_status=exit_status
+    )
+
+
+def find_handler(exception: Exception) -> HOOK_TYPE | None:
+    for handler, condition in _REGISTERED_HOOKS:
+        if not condition(exception):
+            continue
+        return handler
+    return None
+
+
+def _error_handler(
+    *,
+    condition: t.Callable[[Exception], bool],
+    exit_status: int = 1,
 ) -> t.Callable[[_HOOK_SRC_TYPE], HOOK_TYPE]:
     """decorator for excepthooks
 
     register each one, in order, with any relevant "condition"
     """
 
-    def inner_decorator(fn):
+    def inner_decorator(fn: _HOOK_SRC_TYPE) -> HOOK_TYPE:
         @functools.wraps(fn)
-        def wrapped(exception):
+        def wrapped(exception: Exception) -> t.NoReturn:
             hook_result = fn(exception)
             ctx = click.get_current_context()
 
@@ -50,23 +79,44 @@ def wrapped(exception):
 
             ctx.exit(exit_status)
 
-        _REGISTERED_HOOKS.append((wrapped, error_class, condition))
+        _REGISTERED_HOOKS.append((wrapped, condition))
         return wrapped
 
     return inner_decorator
 
 
-def find_handler(exception: Exception) -> HOOK_TYPE | None:
-    for handler, error_class, condition in _REGISTERED_HOOKS:
-        if isinstance(error_class, str):
-            error_class_: type[Exception] = getattr(globus_sdk, error_class)
-            assert issubclass(error_class_, Exception)
-        else:
-            error_class_ = error_class
-
-        if error_class_ is not None and not isinstance(exception, error_class_):
-            continue
-        if condition is not None and not condition(exception):
-            continue
-        return handler
-    return None
+def _build_condition(
+    condition: CONDITION_TYPE | None, error_class: str | type[Exception] | None
+) -> CONDITION_TYPE:
+    inner_condition: CONDITION_TYPE
+
+    if condition is None:
+        if error_class is None:
+            raise ValueError("a hook must specify either condition or error_class")
+
+        def inner_condition(exception: Exception) -> bool:
+            error_class_ = _resolve_error_class(error_class)
+            return isinstance(exception, error_class_)
+
+    elif error_class is None:
+        inner_condition = condition
+
+    else:
+
+        def inner_condition(exception: Exception) -> bool:
+            error_class_ = _resolve_error_class(error_class)
+            return isinstance(exception, error_class_) and condition(exception)
+
+    return inner_condition
+
+
+def _resolve_error_class(error_class: str | type[Exception]) -> type[Exception]:
+    if isinstance(error_class, str):
+        resolved = getattr(globus_sdk, error_class, None)
+        if resolved is None:
+            raise LookupError(f"no such globus_sdk error class '{error_class}'")
+        if not (isinstance(resolved, type) and issubclass(resolved, Exception)):
+            raise ValueError(f"'globus_sdk.{error_class}' is not an error class")
+        return resolved
+    else:
+        return error_class

src/globus_cli/login_manager/manager.py

@@ -231,9 +231,7 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
 
             return wrapper
 
-        # TODO: remove this ignore after a mypy release fixes ParamSpec regressions
-        # see https://github.com/python/mypy/pull/15272 for a candidate fix PR
-        return inner  # type: ignore[return-value]
+        return inner
 
     def _get_client_authorizer(
         self, resource_server: str, *, no_tokens_msg: str | None = None

tox.ini

@@ -66,7 +66,7 @@ base_python =
    python3.11
    python3.10
 deps =
-    mypy==1.5.1
+    mypy==1.7.1
     types-jwt
     types-requests
     types-jmespath