Add list_registered_apis method to the FlowsClient

changelog.d/20260512_120913_8730430+m1yag1_sc_49700_add_list_gras.rst

@@ -0,0 +1,4 @@
+Added
+-----
+
+- Added ``FlowsClient.list_registered_apis()``, which retrieves a list of registered APIs. (:pr:`NUMBER`)

src/globus_sdk/services/flows/__init__.py

@@ -3,6 +3,7 @@
 from .errors import FlowsAPIError
 from .response import (
     IterableFlowsResponse,
+    IterableRegisteredAPIsResponse,
     IterableRunLogsResponse,
     IterableRunsResponse,
 )
@@ -11,8 +12,9 @@
     "FlowsAPIError",
     "FlowsClient",
     "IterableFlowsResponse",
-    "IterableRunsResponse",
+    "IterableRegisteredAPIsResponse",
     "IterableRunLogsResponse",
+    "IterableRunsResponse",
     "SpecificFlowClient",
     "RunActivityNotificationPolicy",
 )

src/globus_sdk/services/flows/client.py

@@ -24,6 +24,7 @@
 from .errors import FlowsAPIError
 from .response import (
     IterableFlowsResponse,
+    IterableRegisteredAPIsResponse,
     IterableRunLogsResponse,
     IterableRunsResponse,
 )
@@ -927,6 +928,58 @@ def get_registered_api(
             f"/registered_apis/{registered_api_id}", query_params=query_params
         )
 
+    @paging.has_paginator(paging.MarkerPaginator, items_key="registered_apis")
+    def list_registered_apis(
+        self,
+        *,
+        filter_roles: str | t.Iterable[str] | MissingType = MISSING,
+        orderby: str | t.Iterable[str] | MissingType = MISSING,
+        marker: str | MissingType = MISSING,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> IterableRegisteredAPIsResponse:
+        """
+        List registered APIs.
+
+        :param filter_roles: Role names to filter results (owner, administrator, viewer)
+        :param orderby: Field and order for sorting results
+        :param marker: Pagination marker for continuing results
+        :param query_params: Any additional parameters to be passed through
+            as query params.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    from globus_sdk import FlowsClient
+
+                    flows = FlowsClient(...)
+                    for api in flows.list_registered_apis(filter_roles="owner"):
+                        print(f"API: {api['name']}")
+
+            .. tab-item:: Paginated Usage
+
+                .. paginatedusage:: list_registered_apis
+
+            .. tab-item:: API Info
+
+                .. extdoclink:: List Registered APIs
+                    :service: flows
+                    :ref: Registered APIs/paths/~1registered_apis/get
+        """
+        query_params = {
+            "filter_roles": commajoin(filter_roles),
+            "orderby": (
+                orderby if isinstance(orderby, (str, MissingType)) else list(orderby)
+            ),
+            "marker": marker,
+            **(query_params or {}),
+        }
+        return IterableRegisteredAPIsResponse(
+            self.get("/registered_apis", query_params=query_params)
+        )
+
 
 class SpecificFlowClient(client.BaseClient):
     r"""

src/globus_sdk/services/flows/response.py

@@ -63,3 +63,24 @@ class IterableRunLogsResponse(response.IterableResponse):
     """
 
     default_iter_key = "entries"
+
+
+class IterableRegisteredAPIsResponse(response.IterableResponse):
+    """
+    An iterable response containing a "registered_apis" array of registered API records.
+
+    This response type is returned by :meth:`FlowsClient.list_registered_apis` and
+    provides iteration over individual registered API objects from a single page of
+    results.
+
+    When iterated over, yields individual registered API dictionaries, where each
+    registered API typically contains:
+
+    - ``id``: UUID of the registered API
+    - ``name``: Display name of the registered API
+    - ``description``: Description of the registered API
+    - ``created_timestamp``: Timestamp of registered API creation
+    - ``updated_timestamp``: Timestamp of last update
+    """
+
+    default_iter_key = "registered_apis"

src/globus_sdk/testing/data/flows/list_registered_apis.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import datetime
+import typing as t
+import uuid
+
+from responses import matchers
+
+from globus_sdk.testing.models import RegisteredResponse, ResponseList, ResponseSet
+
+OWNER_URN = "urn:globus:auth:identity:a1234567-1234-1234-1234-123456789abc"
+
+
+def generate_registered_api_summary(n: int) -> dict[str, t.Any]:
+    """
+    Generate a summary registered API object for list responses.
+
+    :param n: The index number used to generate unique IDs and timestamps
+    """
+    api_id = str(uuid.UUID(int=n))
+    base_time = datetime.datetime.fromisoformat("2024-01-01T00:00:00+00:00")
+    created_timestamp = base_time + datetime.timedelta(days=n)
+    updated_timestamp = created_timestamp + datetime.timedelta(hours=n)
+
+    return {
+        "id": api_id,
+        "name": f"registered-api-{n}",
+        "description": f"Test registered API number {n}",
+        "created_timestamp": created_timestamp.isoformat(),
+        "updated_timestamp": updated_timestamp.isoformat(),
+    }
+
+
+FIRST_REGISTERED_API_ID = str(uuid.UUID(int=0))
+
+RESPONSES = ResponseSet(
+    metadata={"first_registered_api_id": FIRST_REGISTERED_API_ID},
+    default=RegisteredResponse(
+        service="flows",
+        path="/registered_apis",
+        json={
+            "registered_apis": [generate_registered_api_summary(0)],
+            "limit": 1,
+            "has_next_page": False,
+            "marker": None,
+        },
+    ),
+    paginated=ResponseList(
+        RegisteredResponse(
+            service="flows",
+            path="/registered_apis",
+            json={
+                "registered_apis": [
+                    generate_registered_api_summary(i) for i in range(10)
+                ],
+                "limit": 10,
+                "has_next_page": True,
+                "marker": "fake_marker_0",
+            },
+        ),
+        RegisteredResponse(
+            service="flows",
+            path="/registered_apis",
+            json={
+                "registered_apis": [
+                    generate_registered_api_summary(i) for i in range(10, 20)
+                ],
+                "limit": 10,
+                "has_next_page": True,
+                "marker": "fake_marker_1",
+            },
+            match=[matchers.query_param_matcher({"marker": "fake_marker_0"})],
+        ),
+        RegisteredResponse(
+            service="flows",
+            path="/registered_apis",
+            json={
+                "registered_apis": [
+                    generate_registered_api_summary(i) for i in range(20, 25)
+                ],
+                "limit": 5,
+                "has_next_page": False,
+                "marker": None,
+            },
+            match=[matchers.query_param_matcher({"marker": "fake_marker_1"})],
+        ),
+        metadata={
+            "owner_urn": OWNER_URN,
+            "num_pages": 3,
+            "expect_markers": ["fake_marker_0", "fake_marker_1", None],
+            "total_items": 25,
+        },
+    ),
+)

tests/functional/services/flows/test_list_registered_apis.py

@@ -0,0 +1,61 @@
+import urllib.parse
+
+import pytest
+
+from globus_sdk import MISSING
+from globus_sdk.testing import get_last_request, load_response
+
+
+@pytest.mark.parametrize("filter_roles", [MISSING, "owner"])
+@pytest.mark.parametrize("orderby", [MISSING, "name ASC"])
+def test_list_registered_apis_simple(flows_client, filter_roles, orderby):
+    meta = load_response(flows_client.list_registered_apis).metadata
+
+    add_kwargs = {}
+    if filter_roles is not MISSING:
+        add_kwargs["filter_roles"] = filter_roles
+    if orderby is not MISSING:
+        add_kwargs["orderby"] = orderby
+
+    res = flows_client.list_registered_apis(**add_kwargs)
+
+    assert res.http_status == 200
+    # dict-like indexing
+    assert meta["first_registered_api_id"] == res["registered_apis"][0]["id"]
+    # list conversion (using __iter__) and indexing
+    assert meta["first_registered_api_id"] == list(res)[0]["id"]
+
+    req = get_last_request()
+    assert req.body is None
+    parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
+    expect_query_params = {
+        k: [v]
+        for k, v in (
+            ("filter_roles", filter_roles),
+            ("orderby", orderby),
+        )
+        if v is not MISSING
+    }
+    assert parsed_qs == expect_query_params
+
+
+@pytest.mark.parametrize("by_pages", [True, False])
+def test_list_registered_apis_paginated(flows_client, by_pages):
+    meta = load_response(flows_client.list_registered_apis, case="paginated").metadata
+    total_items = meta["total_items"]
+    num_pages = meta["num_pages"]
+    expect_markers = meta["expect_markers"]
+
+    res = flows_client.paginated.list_registered_apis()
+    if by_pages:
+        pages = list(res)
+        assert len(pages) == num_pages
+        for i, page in enumerate(pages):
+            assert page["marker"] == expect_markers[i]
+            if i < num_pages - 1:
+                assert page["has_next_page"] is True
+            else:
+                assert page["has_next_page"] is False
+    else:
+        items = list(res.items())
+        assert len(items) == total_items