Convert globus flows list to use paginated call (#686)

`globus flows list` now uses a paginated call under a PagingWrapper.

To support this, the command has a new argument `--limit`, which is a
norm for any CLI commands which interact with paginated APIs. It
applies to the paginator, not to the API calls. The limit is set, as
with several other commands, to have a floor of 1 and a default of 25.
There is no upper bound set at present.

In order for `-Fjson` output to be usable, we need to set
`json_converter=...` on the formatting call. There is a pre-built
pattern for this for Transfer commands, but it is not portable to
other APIs. Therefore, a new feature is added here to `PagingWrapper`
to generate a value which can be passed here. Ideally the
PagingWrapper could pick this up from the underlying `items_key` in
the paginator being used, and this could be supported by the
formatted_print codepath. However, that would grow the scope of this
change and extends beyond our present needs. (Likely such a change
alters the interface for PagingWrapper and perhaps even allows a
paginator to be directly passed to the formatted_print function.)

Testing for this change copies code from the globus-sdk testsuite to
build out paginated responses. It would be better to share these, but
that requires additional thought and design because the paginated
response set does not trivially fit the existing `globus_sdk._testing`
tools for data sharing.

As additional changes included here, the `globus flows list` command
has been changed in the following ways:
- it passes `orderby=...` as a named parameter, not using
  `query_params`
- it is no longer hidden (the changelog fragment for it is therefore
  reintroduced)

changelog.d/20221013_171820_sirosen_paginated_flow_list.md

@@ -0,0 +1,4 @@
+### Enhancements
+
+* A new command, `globus flows list`, allows users to list Flow objects in
+  Globus Flows

src/globus_cli/commands/flows/__init__.py

@@ -6,7 +6,6 @@
     lazy_subcommands={
         "list": (".list", "list_command"),
     },
-    hidden=True,
 )
 def flows_command():
     """Interact with the Globus Flows service"""

src/globus_cli/commands/flows/list.py

@@ -6,6 +6,7 @@
 from globus_cli.login_manager import LoginManager
 from globus_cli.parsing import command
 from globus_cli.termio import formatted_print
+from globus_cli.utils import PagingWrapper
 
 ROLE_TYPES = ("flow_viewer", "flow_starter", "flow_administrator", "flow_owner")
 
@@ -22,24 +23,37 @@
     help="Filter results based on pattern matching within a subset of fields: "
     "[id, title, subtitle, description, flow_owner, flow_administrators]",
 )
+@click.option(
+    "--limit",
+    default=25,
+    show_default=True,
+    metavar="N",
+    type=click.IntRange(1),
+    help="The maximum number of results to return.",
+)
 @LoginManager.requires_login(LoginManager.FLOWS_RS)
 def list_command(
     login_manager: LoginManager,
     filter_role: str | None,
     filter_fulltext: str | None,
+    limit: int,
 ):
     """
     List flows
     """
     flows_client = login_manager.get_flows_client()
-    # TODO: paginate once path supports pagination
-    #  https://app.shortcut.com/globus/story/18445/add-pagination-back-to-flowsclient-flow-list
-    response = flows_client.list_flows(
-        filter_role=filter_role,
-        filter_fulltext=filter_fulltext,
-        query_params={"orderby": "updated_at DESC"},
+    flow_iterator = PagingWrapper(
+        flows_client.paginated.list_flows(
+            filter_role=filter_role,
+            filter_fulltext=filter_fulltext,
+            orderby="updated_at DESC",
+        ).items(),
+        json_conversion_key="flows",
+        limit=limit,
     )
+
     formatted_print(
-        response["flows"],
+        flow_iterator,
         fields=FLOW_SUMMARY_FORMAT_FIELDS,
+        json_converter=flow_iterator.json_converter,
     )

src/globus_cli/utils.py

@@ -137,10 +137,16 @@ def __getitem__(self, key: str) -> Any:
 
 # wrap to add a `has_next()` method and `limit` param to a naive iterator
 class PagingWrapper:
-    def __init__(self, iterator: Iterator[Any], limit: int | None = None) -> None:
+    def __init__(
+        self,
+        iterator: Iterator[Any],
+        limit: int | None = None,
+        json_conversion_key: str | None = None,
+    ) -> None:
         self.iterator = iterator
         self.next = None
         self.limit = limit
+        self.json_conversion_key = json_conversion_key
         self._step()
 
     def _step(self) -> None:
@@ -160,6 +166,17 @@ def __iter__(self) -> Iterator[Any]:
             yield cur
             yielded += 1
 
+    @property
+    def json_converter(self) -> Callable[[Iterator[Any]], dict[str, list[Any]]]:
+        if self.json_conversion_key is None:
+            raise NotImplementedError("does not support json_converter")
+        key: str = self.json_conversion_key
+
+        def converter(it: Iterator[Any]) -> dict[str, list[Any]]:
+            return {key: list(it)}
+
+        return converter
+
 
 def shlex_process_stream(process_command: click.Command, stream: TextIO) -> None:
     """

tests/functional/flows/conftest.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import datetime
+import uuid
+from typing import Any
+
+import pytest
+from globus_sdk._testing import register_response_set
+from responses import matchers
+
+OWNER_ID = "e061df5a-b7b9-4578-a73b-6d4a4edfd66e"
+
+
+# this builder is mostly copied from globus-sdk tests with minimal changes
+#
+# TODO: find a better way to share the paginated fixture data generation between these
+# two projects
+def generate_hello_world_example_flow(n: int) -> dict[str, Any]:
+    flow_id = str(uuid.UUID(int=n))
+    base_time = datetime.datetime.fromisoformat("2021-10-18T19:19:35.967289+00:00")
+    updated_at = created_at = base_time + datetime.timedelta(days=n)
+    flow_user_scope = (
+        f"https://auth.globus.org/scopes/{flow_id}/"
+        f"flow_{flow_id.replace('-', '_')}_user"
+    )
+
+    return {
+        "action_url": f"https://flows.automate.globus.org/flows/{flow_id}",
+        "administered_by": [],
+        "api_version": "1.0",
+        "created_at": created_at.isoformat(),
+        "created_by": f"urn:globus:auth:identity:{OWNER_ID}",
+        "definition": {
+            "StartAt": "HelloWorld",
+            "States": {
+                "HelloWorld": {
+                    "ActionScope": (
+                        "https://auth.globus.org/scopes/actions.globus.org/hello_world"
+                    ),
+                    "ActionUrl": "https://actions.globus.org/hello_world",
+                    "End": True,
+                    "Parameters": {"echo_string": "Hello, World."},
+                    "ResultPath": "$.Result",
+                    "Type": "Action",
+                }
+            },
+        },
+        "description": "A simple Flow...",
+        "flow_administrators": [],
+        "flow_owner": f"urn:globus:auth:identity:{OWNER_ID}",
+        "flow_starters": [],
+        "flow_url": f"https://flows.automate.globus.org/flows/{flow_id}",
+        "flow_viewers": [],
+        "globus_auth_scope": flow_user_scope,
+        "globus_auth_username": f"{flow_id}@clients.auth.globus.org",
+        "id": str(flow_id),
+        "input_schema": {
+            "additionalProperties": False,
+            "properties": {
+                "echo_string": {"description": "The string to echo", "type": "string"},
+                "sleep_time": {"type": "integer"},
+            },
+            "required": ["echo_string", "sleep_time"],
+            "type": "object",
+        },
+        "keywords": [],
+        "log_supported": True,
+        "principal_urn": f"urn:globus:auth:identity:{flow_id}",
+        "runnable_by": [],
+        "subtitle": "",
+        "synchronous": False,
+        "title": f"Hello, World (Example {n})",
+        "types": ["Action"],
+        "updated_at": updated_at.isoformat(),
+        "user_role": "flow_viewer",
+        "visible_to": [],
+    }
+
+
+@pytest.fixture(autouse=True, scope="session")
+def setup_paginated_responses() -> None:
+    register_response_set(
+        "flows_list_paginated",
+        {
+            "page0": dict(
+                service="flows",
+                path="/flows",
+                json={
+                    "flows": [generate_hello_world_example_flow(i) for i in range(20)],
+                    "limit": 20,
+                    "has_next_page": True,
+                    "marker": "fake_marker_0",
+                },
+                match=[matchers.query_param_matcher({"orderby": "updated_at DESC"})],
+            ),
+            "page1": dict(
+                service="flows",
+                path="/flows",
+                json={
+                    "flows": [
+                        generate_hello_world_example_flow(i) for i in range(20, 40)
+                    ],
+                    "limit": 20,
+                    "has_next_page": True,
+                    "marker": "fake_marker_1",
+                },
+                match=[
+                    matchers.query_param_matcher(
+                        {"orderby": "updated_at DESC", "marker": "fake_marker_0"}
+                    )
+                ],
+            ),
+            "page2": dict(
+                service="flows",
+                path="/flows",
+                json={
+                    "flows": [
+                        generate_hello_world_example_flow(i) for i in range(40, 60)
+                    ],
+                    "limit": 20,
+                    "has_next_page": False,
+                    "marker": None,
+                },
+                match=[
+                    matchers.query_param_matcher(
+                        {"orderby": "updated_at DESC", "marker": "fake_marker_1"}
+                    )
+                ],
+            ),
+            "auth_get_identities": dict(
+                service="auth",
+                path="/v2/api/identities",
+                json={
+                    "identities": [
+                        {
+                            "username": "shrek@globus.org",
+                            "name": "Shrek by William Steig",
+                            "id": OWNER_ID,
+                            "identity_provider": "c8abac57-560c-46c8-b386-f116ed8793d5",
+                            "organization": (
+                                "Fairytales Whose Movie Adaptations Diverge "
+                                "Significantly From Their Source Material"
+                            ),
+                            "status": "used",
+                            "email": "shrek@globus.org",
+                        }
+                    ]
+                },
+            ),
+        },
+        metadata={
+            "owner_id": OWNER_ID,
+            "flow_owner": "shrek@globus.org",
+            "num_pages": 3,
+            "expect_markers": ["fake_marker_0", "fake_marker_1", None],
+            "total_items": 60,
+        },
+    )

tests/functional/flows/test_list_flows.py

@@ -1,4 +1,5 @@
 import json
+import uuid
 
 from globus_sdk._testing import load_response_set
 
@@ -59,3 +60,20 @@ def test_list_flows_filter_fulltext(run_line):
 
     result = run_line("globus flows list --filter-fulltext Fairytale")
     assert result.output == expected
+
+
+def test_list_flows_paginated_response(run_line):
+    meta = load_response_set("flows_list_paginated").metadata
+
+    result = run_line("globus flows list --limit 1000")
+    output_lines = result.output.split("\n")[:-1]  # trim the final newline/empty str
+    assert len(output_lines) == meta["total_items"] + 2
+
+    for i, line in enumerate(output_lines[2:]):
+        row = line.split(" | ")
+        assert row[0] == str(uuid.UUID(int=i))
+        # rstrip because this column may be right-padded to align
+        #    Hello, World (Example {1})   <-- trailing space
+        #    Hello, World (Example {10})  <-- no trailing space
+        assert row[1].rstrip() == f"Hello, World (Example {i})"
+        assert row[2] == meta["flow_owner"]