Split up error hooks to be more modular, making ordering explicit

src/globus_cli/exception_handling/hooks/__init__.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import typing as t
+
+from ..registry import DeclaredHook, register_hook
+from .auth_requirements import (
+    consent_required_hook,
+    handle_internal_auth_requirements,
+    missing_login_error_hook,
+    session_hook,
+)
+from .authapi_hooks import (
+    authapi_hook,
+    authapi_unauthenticated_hook,
+    invalidrefresh_hook,
+)
+from .endpoint_types import wrong_endpoint_type_error_hook
+from .flows_hooks import (
+    flows_error_hook,
+    flows_validation_error_hook,
+    handle_flows_gare,
+)
+from .generic_hooks import (
+    globus_error_hook,
+    globusapi_hook,
+    json_error_handler,
+    null_data_error_handler,
+)
+from .search_hooks import searchapi_hook, searchapi_validationerror_hook
+from .transfer_hooks import transfer_unauthenticated_hook, transferapi_hook
+
+
+def register_all_hooks() -> None:
+    """
+    Load and register all hook functions.
+    """
+    for hook in _sort_all_hooks():
+        register_hook(hook)
+
+
+def _sort_all_hooks() -> t.Iterable[DeclaredHook[t.Any]]:
+    """
+    Iterate over all hooks in priority order.
+    """
+    # format as a list of lists for readability
+    sorted_hook_collections: list[list[DeclaredHook[t.Any]]] = [
+        # first, the generic hooks which filter out conditions
+        # running 'null data' first ensures that every other
+        # hook can assume that there is a JSON body
+        # and running json_error_handler at the start means we know (in the
+        # following hooks) that the output format is not JSON
+        [null_data_error_handler, json_error_handler],
+        # next, authn and session requirements, from most specific to most general
+        [handle_internal_auth_requirements, handle_flows_gare],
+        [session_hook, consent_required_hook],
+        # CLI internal error types, which cannot be confused with external causes
+        [missing_login_error_hook, wrong_endpoint_type_error_hook],
+        # service-specific hooks uncaptured by earlier checks
+        # each service has internal precedence ordering, but the collections could
+        # probably be put in any order
+        [authapi_unauthenticated_hook, invalidrefresh_hook, authapi_hook],
+        [transfer_unauthenticated_hook, transferapi_hook],
+        [flows_validation_error_hook, flows_error_hook],
+        [searchapi_validationerror_hook, searchapi_hook],
+        # finally, the catch-all hooks
+        [globusapi_hook, globus_error_hook],
+    ]
+
+    for hook_collection in sorted_hook_collections:
+        yield from hook_collection

src/globus_cli/exception_handling/hooks/auth_requirements.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import click
+import globus_sdk
+
+from globus_cli.login_manager import MissingLoginError
+from globus_cli.utils import CLIAuthRequirementsError
+
+from ..messages import (
+    DEFAULT_CONSENT_REAUTH_MESSAGE,
+    DEFAULT_SESSION_REAUTH_MESSAGE,
+    emit_consent_required_message,
+    emit_message_for_gare,
+    emit_session_update_message,
+)
+from ..registry import error_handler, sdk_error_handler
+
+
+@error_handler(error_class=CLIAuthRequirementsError, exit_status=4)
+def handle_internal_auth_requirements(
+    exception: CLIAuthRequirementsError,
+) -> int | None:
+    gare = exception.gare
+    if not gare:
+        msg = "Fatal Error: Unsupported internal auth requirements error!"
+        click.secho(msg, bold=True, fg="red")
+        return 255
+
+    emit_message_for_gare(gare, exception.message)
+
+    if exception.epilog:
+        click.echo("\n* * *\n")
+        click.echo(exception.epilog)
+
+    return None
+
+
+@sdk_error_handler(
+    condition=lambda err: bool(err.info.authorization_parameters), exit_status=4
+)
+def session_hook(exception: globus_sdk.GlobusAPIError) -> None:
+    """
+    Expects an exception with a valid authorization_paramaters info field.
+    """
+    message = exception.info.authorization_parameters.session_message
+    if message:
+        message = f"{DEFAULT_SESSION_REAUTH_MESSAGE}\nmessage: {message}"
+    else:
+        message = DEFAULT_SESSION_REAUTH_MESSAGE
+
+    emit_session_update_message(
+        identities=exception.info.authorization_parameters.session_required_identities,
+        domains=exception.info.authorization_parameters.session_required_single_domain,
+        policies=exception.info.authorization_parameters.session_required_policies,
+        message=message,
+    )
+    return None
+
+
+@sdk_error_handler(condition=lambda err: bool(err.info.consent_required), exit_status=4)
+def consent_required_hook(exception: globus_sdk.GlobusAPIError) -> int | None:
+    """
+    Expects an exception with a required_scopes field in its raw_json.
+    """
+    if not exception.info.consent_required.required_scopes:
+        msg = "Fatal Error: ConsentRequired but no required_scopes!"
+        click.secho(msg, bold=True, fg="red")
+        return 255
+
+    # specialized message for data_access errors
+    # otherwise, use more generic phrasing
+    if exception.message == "Missing required data_access consent":
+        message = (
+            "The collection you are trying to access data on requires you to "
+            "grant consent for the Globus CLI to access it."
+        )
+    else:
+        message = f"{DEFAULT_CONSENT_REAUTH_MESSAGE}\nmessage: {exception.message}"
+
+    emit_consent_required_message(
+        required_scopes=exception.info.consent_required.required_scopes, message=message
+    )
+    return None
+
+
+@error_handler(error_class=MissingLoginError, exit_status=4)
+def missing_login_error_hook(exception: MissingLoginError) -> None:
+    click.echo(
+        click.style("MissingLoginError: ", fg="yellow") + exception.message,
+        err=True,
+    )

src/globus_cli/exception_handling/hooks/authapi_hooks.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import globus_sdk
+
+from globus_cli.termio import PrintableErrorField, write_error_info
+
+from ..messages import emit_unauthorized_message
+from ..registry import sdk_error_handler
+
+
+@sdk_error_handler(
+    error_class="AuthAPIError", condition=lambda err: err.code == "UNAUTHORIZED"
+)
+def authapi_unauthenticated_hook(exception: globus_sdk.AuthAPIError) -> None:
+    emit_unauthorized_message()
+
+
+@sdk_error_handler(
+    error_class="AuthAPIError",
+    condition=lambda err: err.message == "invalid_grant",
+)
+def invalidrefresh_hook(exception: globus_sdk.AuthAPIError) -> None:
+    emit_unauthorized_message()
+
+
+@sdk_error_handler(error_class="AuthAPIError")
+def authapi_hook(exception: globus_sdk.AuthAPIError) -> None:
+    write_error_info(
+        "Auth API Error",
+        [
+            PrintableErrorField("HTTP status", exception.http_status),
+            PrintableErrorField("code", exception.code),
+            PrintableErrorField("message", exception.message, multiline=True),
+        ],
+    )

src/globus_cli/exception_handling/hooks/endpoint_types.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import click
+
+from globus_cli.endpointish import WrongEntityTypeError
+
+from ..registry import error_handler
+
+
+@error_handler(error_class=WrongEntityTypeError, exit_status=3)
+def wrong_endpoint_type_error_hook(exception: WrongEntityTypeError) -> None:
+    msg = exception.expected_message + "\n" + exception.actual_message + "\n\n"
+    click.secho(msg, fg="yellow", err=True)
+
+    should_use = exception.should_use_command()
+    if should_use:
+        click.echo(
+            "Please run the following command instead:\n\n"
+            f"    {should_use} {exception.endpoint_id}\n",
+            err=True,
+        )
+    else:
+        msg = "This operation is not supported on objects of this type."
+        click.secho(msg, fg="red", bold=True, err=True)