Added globus flows registered-api show command

changelog.d/20260507_153740_8730430+m1yag1_sc_49699_add_gra_show_command.md

@@ -0,0 +1,3 @@
+### Enhancements
+
+* Added `globus flows registered-api show` command to display details of a registered API

pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
 ]
 requires-python = ">=3.9"
 dependencies = [
-    "globus-sdk==4.4.0",
+    "globus-sdk==4.6.0",
     "click>=8.1.4,<8.4",
     "jmespath==1.1.0",
     "packaging>=17.0",

src/globus_cli/commands/flows/__init__.py

@@ -13,6 +13,8 @@
         "start": (".start", "start_command"),
         # "run" is a subgroup of commands.
         "run": (".run", "run_command"),
+        # "registered-api" is a subgroup of commands.
+        "registered-api": (".registered_api", "registered_api_command"),
     },
 )
 def flows_command() -> None:

src/globus_cli/commands/flows/_fields.py

@@ -111,3 +111,57 @@ def flow_run_format_fields(
         Field("Run Managers", "run_managers", formatter=csv_principal_list),
         Field("Run Monitors", "run_monitors", formatter=csv_principal_list),
     ]
+
+
+class RegisteredAPIPrincipalFormatter(PrincipalURNFormatter):
+    """A principal formatter which pre-registers all principals for a registered API."""
+
+    def __init__(
+        self, auth_client: globus_sdk.AuthClient, registered_api: dict[str, t.Any]
+    ) -> None:
+        super().__init__(auth_client)
+        roles = registered_api.get("roles", {})
+        self.add_items(*roles.get("owners", ()))
+        self.add_items(*roles.get("administrators", ()))
+        self.add_items(*roles.get("viewers", ()))
+
+
+def registered_api_format_fields(
+    auth_client: globus_sdk.AuthClient,
+    registered_api: dict[str, t.Any],
+) -> list[Field]:
+    """
+    The standard list of fields to render for a registered API resource.
+
+    :param auth_client: An AuthClient, used to resolve principal URNs.
+    :param registered_api: The registered API resource, used to pre-register
+        principals.
+    """
+    principal = RegisteredAPIPrincipalFormatter(auth_client, registered_api)
+    csv_principal_list = formatters.ArrayFormatter(
+        element_formatter=principal,
+        delimiter=", ",
+    )
+
+    return [
+        Field("Registered API ID", "id"),
+        Field("Name", "name"),
+        Field("Description", "description"),
+        Field("Status", "status"),
+        Field("Subscription ID", "subscription_id"),
+        Field("Created At", "created_timestamp", formatter=formatters.Date),
+        Field("Updated At", "updated_timestamp", formatter=formatters.Date),
+        Field("Edited At", "edited_timestamp", formatter=formatters.Date),
+        Field(
+            "Scheduled Deletion",
+            "scheduled_deletion_timestamp",
+            formatter=formatters.Date,
+        ),
+        Field("Owners", "roles.owners", formatter=csv_principal_list),
+        Field("Administrators", "roles.administrators", formatter=csv_principal_list),
+        Field("Viewers", "roles.viewers", formatter=csv_principal_list),
+        Field("Target Type", "target.type"),
+        Field("OpenAPI Version", "target.openapi_version"),
+        Field("Destination Method", "target.destination.method"),
+        Field("Destination URL", "target.destination.url"),
+    ]

src/globus_cli/commands/flows/registered_api/__init__.py

@@ -0,0 +1,11 @@
+from globus_cli.parsing import group
+
+
+@group(
+    "registered-api",
+    lazy_subcommands={
+        "show": (".show", "show_command"),
+    },
+)
+def registered_api_command() -> None:
+    """Interact with registered APIs in the Globus Flows service."""

src/globus_cli/commands/flows/registered_api/show.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import uuid
+
+import click
+
+from globus_cli.commands.flows._fields import registered_api_format_fields
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import command
+from globus_cli.termio import display
+
+
+@command("show")
+@click.argument("registered_api_id", type=click.UUID)
+@LoginManager.requires_login("auth", "flows")
+def show_command(login_manager: LoginManager, *, registered_api_id: uuid.UUID) -> None:
+    """
+    Show a registered API.
+    """
+    flows_client = login_manager.get_flows_client()
+
+    res = flows_client.get_registered_api(registered_api_id)
+
+    auth_client = login_manager.get_auth_client()
+    fields = registered_api_format_fields(auth_client, res.data)
+
+    display(res, fields=fields, text_mode=display.RECORD)

tests/functional/flows/conftest.py

@@ -274,3 +274,31 @@ def setup_custom_orderby_response() -> None:
             "total_items": 26,
         },
     )
+
+
+@pytest.fixture
+def load_identities_for_registered_api(get_identities_mocker):
+    """
+    Callable fixture.
+    Load identities for a provided registered API response object, configuring
+    auth.get_identities to return unique usernames for each identity associated with the
+    registered API.
+
+    Returns an identity pool object to facilitate lookup of username by principal urn.
+    """
+
+    def _load_identities_for_registered_api(
+        registered_api: dict[str, t.Any],
+    ) -> FlowsIdentityPool:
+        principals: set[str] = set()
+        roles = registered_api.get("roles", {})
+        principals.update(set(roles.get("owners", [])))
+        principals.update(set(roles.get("administrators", [])))
+        principals.update(set(roles.get("viewers", [])))
+
+        pool = FlowsIdentityPool(principals)
+
+        get_identities_mocker.configure(pool.create_get_identities_documents())
+        return pool
+
+    return _load_identities_for_registered_api

tests/functional/flows/test_show_registered_api.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import re
+
+from globus_sdk.testing import load_response
+
+
+def test_show_registered_api_text_output(run_line, load_identities_for_registered_api):
+    loaded_response = load_response("flows.get_registered_api")
+    response, meta = loaded_response.json, loaded_response.metadata
+
+    registered_api_id = meta["registered_api_id"]
+
+    pool = load_identities_for_registered_api(response)
+
+    result = run_line(f"globus flows registered-api show {registered_api_id}")
+
+    # all fields present
+    for fieldname in (
+        "Registered API ID",
+        "Name",
+        "Description",
+        "Status",
+        "Subscription ID",
+        "Created At",
+        "Updated At",
+        "Owners",
+        "Administrators",
+        "Viewers",
+        "Target Type",
+        "OpenAPI Version",
+        "Destination Method",
+        "Destination URL",
+    ):
+        assert fieldname in result.output
+
+    # Verify principal resolution
+    roles = response.get("roles", {})
+    assert_usernames(result, pool, "Owners", roles.get("owners", []))
+    assert_usernames(result, pool, "Administrators", roles.get("administrators", []))
+    assert_usernames(result, pool, "Viewers", roles.get("viewers", []))
+
+
+def assert_usernames(result, pool, field_name, principals):
+    expected_usernames = {pool.get_username(principal) for principal in principals}
+
+    output_value = _get_output_value(field_name, result.output)
+    output_usernames = {x.strip() for x in output_value.split(",")}
+    assert expected_usernames == output_usernames
+
+
+def _get_output_value(name, output):
+    """
+    Return the value for a specified field from the output of a command.
+    """
+    match = re.search(
+        rf"^{re.escape(name)}:[^\S\n\r]+(?P<value>.*)$", output, flags=re.M
+    )
+    assert match is not None
+    return match.group("value")