Add update command to modify a registered API (#17)

Author: m1yag1

changelog.d/20260203_163857_8730430+m1yag1_sc_47176_implement_update_cli_command.rst

@@ -0,0 +1,5 @@
+
+Added
+-----
+
+*   Add ``update`` CLI command to modify a registered API by ID.

src/globus_registered_api/cli.py

@@ -6,8 +6,11 @@
 
 import json
 import os
+import pathlib
 import sys
 from collections.abc import Iterable
+from typing import Any
+from uuid import UUID
 
 import click
 from globus_sdk import AuthClient
@@ -236,6 +239,116 @@ def get_registered_api(ctx: click.Context, registered_api_id: str, format: str)
         click.echo(f"Updated:     {res['updated_timestamp']}")
 
 
+@cli.command("update")
+@click.argument("registered_api_id", type=click.UUID)
+@click.option("--name", help="Update the name of the registered API")
+@click.option("--description", help="Update the description of the registered API")
+@click.option(
+    "--owner",
+    "owners",
+    multiple=True,
+    help="Set owner URN (can specify multiple, can only be set by owners)",
+)
+@click.option(
+    "--administrator",
+    "administrators",
+    multiple=True,
+    help="Set administrator URN (can specify multiple, can only be set by owners)",
+)
+@click.option(
+    "--viewer",
+    "viewers",
+    multiple=True,
+    help="Set viewer URN (can specify multiple, can only be set by owners and administrators)",
+)
+@click.option(
+    "--no-administrators",
+    is_flag=True,
+    help="Clear all administrators (can only be set by owners)",
+)
+@click.option(
+    "--no-viewers",
+    is_flag=True,
+    help="Clear all viewers (can only be set by owners and administrators)",
+)
+@click.option(
+    "--target-file",
+    type=click.Path(exists=True, dir_okay=False, path_type=pathlib.Path),
+    help="Path to JSON file containing target definition",
+)
+@click.option("--format", type=click.Choice(["json", "text"]), default="text")
+@click.pass_context
+def update_registered_api(
+    ctx: click.Context,
+    registered_api_id: UUID,
+    name: str | None,
+    description: str | None,
+    owners: tuple[str, ...],
+    administrators: tuple[str, ...],
+    viewers: tuple[str, ...],
+    no_administrators: bool,
+    no_viewers: bool,
+    target_file: pathlib.Path | None,
+    format: str,
+) -> None:
+    """
+    Update a registered API by ID.
+
+    Use this command to modify the name, description, roles, or target
+    definition of an existing registered API.
+    """
+    app: UserApp | ClientApp = ctx.obj
+    flows_client = _create_flows_client(app)
+
+    # Validate mutually exclusive options
+    if administrators and no_administrators:
+        raise click.UsageError(
+            "--administrator and --no-administrators cannot be used together"
+        )
+    if viewers and no_viewers:
+        raise click.UsageError(
+            "--viewer and --no-viewers cannot be used together"
+        )
+
+    request: dict[str, Any] = {}
+    if name is not None:
+        request["name"] = name
+    if description is not None:
+        request["description"] = description
+    if owners:
+        request["owners"] = list(set(owners))
+    if no_administrators:
+        request["administrators"] = []
+    elif administrators:
+        request["administrators"] = list(set(administrators))
+    if no_viewers:
+        request["viewers"] = []
+    elif viewers:
+        request["viewers"] = list(set(viewers))
+    if target_file is not None:
+        try:
+            request["target"] = json.loads(target_file.read_text())
+        except json.JSONDecodeError as e:
+            raise click.UsageError(f"Invalid JSON in target file: {e}")
+        except UnicodeDecodeError as e:
+            raise click.UsageError(f"Unable to read target file: {e}")
+
+    res = flows_client.update_registered_api(registered_api_id, **request)
+
+    if format == "json":
+        click.echo(json.dumps(res.data, indent=2))
+    else:
+        click.echo(f"ID:             {res['id']}")
+        click.echo(f"Name:           {res['name']}")
+        click.echo(f"Description:    {res['description']}")
+        click.echo(f"Owners:         {res['roles']['owners']}")
+        click.echo(f"Administrators: {res['roles']['administrators']}")
+        click.echo(f"Viewers:        {res['roles']['viewers']}")
+        click.echo(f"Created:        {res['created_timestamp']}")
+        if res.get("edited_timestamp"):
+            click.echo(f"Edited:         {res['edited_timestamp']}")
+
+
 # --- willdelete command group ---
 
 

src/globus_registered_api/extended_flows_client.py

@@ -79,3 +79,46 @@ def get_registered_api(
         :return: Response containing the registered API details
         """
         return self.get(f"/registered_apis/{registered_api_id}")
+
+    def update_registered_api(
+        self,
+        registered_api_id: str | uuid.UUID,
+        *,
+        name: str | None = None,
+        description: str | None = None,
+        owners: list[str] | None = None,
+        administrators: list[str] | None = None,
+        viewers: list[str] | None = None,
+        target: dict[str, t.Any] | None = None,
+    ) -> GlobusHTTPResponse:
+        """
+        Update a registered API by ID.
+
+        :param registered_api_id: The ID of the registered API to update
+        :param name: New name for the registered API
+        :param description: New description for the registered API
+        :param owners: List of owner URNs (replaces existing owners)
+        :param administrators: List of administrator URNs (replaces existing)
+        :param viewers: List of viewer URNs (replaces existing)
+        :param target: Target definition dict
+        :return: Response containing the updated registered API
+        """
+        body: dict[str, t.Any] = _filter_nones(
+            {
+                "name": name,
+                "description": description,
+                "target": target,
+            }
+        )
+
+        roles = _filter_nones(
+            {
+                "owners": owners,
+                "administrators": administrators,
+                "viewers": viewers,
+            }
+        )
+        if roles:
+            body["roles"] = roles
+
+        return self.patch(f"/registered_apis/{registered_api_id}", data=body)

tests/conftest.py

@@ -16,6 +16,9 @@
 GET_REGISTERED_API_URL = re.compile(
     r"https://.*flows.*\.globus\.org/registered_apis/[a-f0-9-]+"
 )
+UPDATE_REGISTERED_API_URL = re.compile(
+    r"https://.*flows.*\.globus\.org/registered_apis/[a-f0-9-]+"
+)
 
 
 @pytest.fixture(autouse=True)

tests/test_extended_flows_client.py

@@ -3,17 +3,22 @@
 # Copyright 2025 Globus <support@globus.org>
 # SPDX-License-Identifier: Apache-2.0
 
+import json
 import uuid
 
 import pytest
 import responses
-from conftest import GET_REGISTERED_API_URL
-from conftest import LIST_REGISTERED_APIS_URL
+from conftest import (
+    GET_REGISTERED_API_URL,
+    LIST_REGISTERED_APIS_URL,
+    UPDATE_REGISTERED_API_URL,
+)
 from globus_sdk import GlobusHTTPResponse
 
 from globus_registered_api.extended_flows_client import ExtendedFlowsClient
 
 
+
 @pytest.fixture
 def client():
     return ExtendedFlowsClient()
@@ -179,3 +184,139 @@ def test_get_registered_api(client):
     assert response["name"] == "Test API"
     assert response["description"] == "A test API"
     assert f"/registered_apis/{api_id}" in responses.calls[0].request.url
+
+
+def test_update_registered_api_basic(client):
+    api_id = uuid.uuid4()
+    responses.add(
+        responses.PATCH,
+        UPDATE_REGISTERED_API_URL,
+        json={
+            "id": str(api_id),
+            "name": "Updated API",
+            "description": "Updated description",
+            "roles": {
+                "owners": ["urn:globus:auth:identity:user1"],
+                "administrators": [],
+                "viewers": [],
+            },
+            "created_timestamp": "2025-01-01T00:00:00+00:00",
+            "edited_timestamp": None,
+        },
+    )
+
+    response = client.update_registered_api(api_id, name="Updated API")
+
+    assert isinstance(response, GlobusHTTPResponse)
+    assert response["name"] == "Updated API"
+    assert f"/registered_apis/{api_id}" in responses.calls[0].request.url
+    assert responses.calls[0].request.method == "PATCH"
+
+
+def test_update_registered_api_with_description(client):
+    api_id = uuid.uuid4()
+    responses.add(
+        responses.PATCH,
+        UPDATE_REGISTERED_API_URL,
+        json={
+            "id": str(api_id),
+            "name": "Test API",
+            "description": "New description",
+            "roles": {
+                "owners": ["urn:globus:auth:identity:user1"],
+                "administrators": [],
+                "viewers": [],
+            },
+            "created_timestamp": "2025-01-01T00:00:00+00:00",
+            "edited_timestamp": None,
+        },
+    )
+
+    response = client.update_registered_api(api_id, description="New description")
+
+    assert response["description"] == "New description"
+
+
+def test_update_registered_api_with_roles(client):
+    api_id = uuid.uuid4()
+    new_owners = ["urn:globus:auth:identity:user1", "urn:globus:auth:identity:user2"]
+    responses.add(
+        responses.PATCH,
+        UPDATE_REGISTERED_API_URL,
+        json={
+            "id": str(api_id),
+            "name": "Test API",
+            "description": "Test",
+            "roles": {
+                "owners": new_owners,
+                "administrators": [],
+                "viewers": [],
+            },
+            "created_timestamp": "2025-01-01T00:00:00+00:00",
+            "edited_timestamp": None,
+        },
+    )
+
+    response = client.update_registered_api(api_id, owners=new_owners)
+
+    assert response["roles"]["owners"] == new_owners
+
+
+def test_update_registered_api_with_target(client):
+    api_id = uuid.uuid4()
+    target = {
+        "type": "openapi",
+        "openapi_version": "3.1",
+        "destination": {"method": "get", "url": "https://example.com/api"},
+        "specification": {"operationId": "test-op", "responses": {}},
+    }
+    responses.add(
+        responses.PATCH,
+        UPDATE_REGISTERED_API_URL,
+        json={
+            "id": str(api_id),
+            "name": "Test API",
+            "description": "Test",
+            "roles": {
+                "owners": ["urn:globus:auth:identity:user1"],
+                "administrators": [],
+                "viewers": [],
+            },
+            "target": target,
+            "created_timestamp": "2025-01-01T00:00:00+00:00",
+            "edited_timestamp": "2025-01-02T00:00:00+00:00",
+        },
+    )
+
+    response = client.update_registered_api(api_id, target=target)
+
+    assert response["target"] == target
+
+
+def test_update_registered_api_omitted_params_not_in_request(client):
+    api_id = uuid.uuid4()
+    responses.add(
+        responses.PATCH,
+        UPDATE_REGISTERED_API_URL,
+        json={
+            "id": str(api_id),
+            "name": "Updated Name",
+            "description": "Original description",
+            "roles": {
+                "owners": ["urn:globus:auth:identity:user1"],
+                "administrators": [],
+                "viewers": [],
+            },
+            "created_timestamp": "2025-01-01T00:00:00+00:00",
+            "edited_timestamp": None,
+        },
+    )
+
+    client.update_registered_api(api_id, name="Updated Name")
+
+    request_body = json.loads(responses.calls[0].request.body)
+    # Only 'name' should be in the request, not None values for omitted params
+    assert "name" in request_body
+    assert "description" not in request_body
+    assert "target" not in request_body
+    assert "roles" not in request_body