Merge branch 'main' into issue969

Author: sgillies

CHANGES.txt

@@ -1,8 +1,10 @@
 2.1.0 (TBD)
 
 Added:
-- Support for catalog source publishing stages (#977) and time range types ()
-  have been added to subscription_request.catalog_source.
+- Support for catalog source publishing stages (#977) and time range types
+  (#978) have been added to subscription_request.catalog_source.
+- Add the option to get Planetary Variable subscription results as a CSV file
+  (#981).
 - A subscription_request.planetary_variable_source function has been added
   (#976).
 - The subscription_request.build_request function has a new option to clip to

planet/cli/subscriptions.py

@@ -162,10 +162,13 @@ async def get_subscription_cmd(ctx, subscription_id, pretty):
                        "success"]),
     multiple=True,
     default=None,
-    callback=lambda ctx,
-    param,
-    value: set(value),
+    callback=(lambda ctx, param, value: set(value)),
     help="Select subscription results in one or more states. Default: all.")
+@click.option('--csv',
+              'csv_flag',
+              is_flag=True,
+              default=False,
+              help="Get subscription results as comma-separated fields.")
 @limit
 # TODO: the following 3 options.
 # –created: timestamp instant or range.
@@ -178,13 +181,39 @@ async def list_subscription_results_cmd(ctx,
                                         subscription_id,
                                         pretty,
                                         status,
+                                        csv_flag,
                                         limit):
-    """Gets results of a subscription and prints the API response."""
+    """Print the results of a subscription to stdout.
+
+    The output of this command is a sequence of JSON objects (the
+    default) or a sequence of comma-separated fields (when the --csv
+    option is used), one result per line.
+
+    Examples:
+
+    \b
+        planet subscriptions results SUBSCRIPTION_ID --status=success --limit 10
+
+    Where SUBSCRIPTION_ID is the unique identifier for a subscription,
+    this prints the last 10 successfully delivered results for that
+    subscription as JSON objects.
+
+    \b
+        planet subscriptions results SUBSCRIPTION_ID --limit 0 --csv > results.csv
+
+    Prints all results for a subscription and saves them to a CSV file.
+    """
     async with subscriptions_client(ctx) as client:
-        async for result in client.get_results(subscription_id,
-                                               status=status,
-                                               limit=limit):
-            echo_json(result, pretty)
+        if csv_flag:
+            async for result in client.get_results_csv(subscription_id,
+                                                       status=status,
+                                                       limit=limit):
+                click.echo(result)
+        else:
+            async for result in client.get_results(subscription_id,
+                                                   status=status,
+                                                   limit=limit):
+                echo_json(result, pretty)
 
 
 @subscriptions.command()  # type: ignore

planet/clients/subscriptions.py

@@ -1,7 +1,7 @@
 """Planet Subscriptions API Python client."""
 
 import logging
-from typing import AsyncIterator, Optional, Set
+from typing import AsyncIterator, Literal, Optional, Sequence
 
 from planet.exceptions import APIError, ClientError
 from planet.http import Session
@@ -58,7 +58,7 @@ def __init__(self,
             self._base_url = self._base_url[:-1]
 
     async def list_subscriptions(self,
-                                 status: Optional[Set[str]] = None,
+                                 status: Optional[Sequence[str]] = None,
                                  limit: int = 100) -> AsyncIterator[dict]:
         """Iterate over list of account subscriptions with optional filtering.
 
@@ -216,16 +216,21 @@ async def get_subscription(self, subscription_id: str) -> dict:
 
     async def get_results(self,
                           subscription_id: str,
-                          status: Optional[Set[str]] = None,
+                          status: Optional[Sequence[Literal[
+                              "created",
+                              "queued",
+                              "processing",
+                              "failed",
+                              "success"]]] = None,
                           limit: int = 100) -> AsyncIterator[dict]:
         """Iterate over results of a Subscription.
 
-        Note:
+        Notes:
             The name of this method is based on the API's method name. This
             method provides iteration over results, it does not get a
             single result description or return a list of descriptions.
 
-        Args:
+        Parameters:
             subscription_id (str): id of a subscription.
             status (Set[str]): pass result with status in this set,
                 filter out results with status not in this set.
@@ -252,7 +257,6 @@ class _ResultsPager(Paged):
             resp = await self._session.request(method='GET',
                                                url=url,
                                                params=params)
-
             async for sub in _ResultsPager(resp,
                                            self._session.request,
                                            limit=limit):
@@ -263,3 +267,41 @@ class _ResultsPager(Paged):
             raise
         except ClientError:  # pragma: no cover
             raise
+
+    async def get_results_csv(
+        self,
+        subscription_id: str,
+        status: Optional[Sequence[Literal["created",
+                                          "queued",
+                                          "processing",
+                                          "failed",
+                                          "success"]]] = None,
+        limit: int = 100,
+    ) -> AsyncIterator[str]:
+        """Iterate over rows of results CSV for a Subscription.
+
+        Parameters:
+            subscription_id (str): id of a subscription.
+            status (Set[str]): pass result with status in this set,
+                filter out results with status not in this set.
+            TODO: created, updated, completed, user_id
+
+        Yields:
+            str: a row from a CSV file.
+
+        Raises:
+            APIError: on an API server error.
+            ClientError: on a client error.
+        """
+        url = f'{self._base_url}/{subscription_id}/results'
+        params = {'status': [val for val in status or {}], 'format': 'csv'}
+
+        # Note: retries are not implemented yet. This project has
+        # retry logic for HTTP requests, but does not handle errors
+        # during streaming. We may want to consider a retry decorator
+        # for this entire method a la stamina:
+        # https://github.com/hynek/stamina.
+        async with self._session._client.stream('GET', url,
+                                                params=params) as response:
+            async for line in response.aiter_lines():
+                yield line

planet/subscription_request.py

@@ -297,8 +297,8 @@ def planetary_variable_source(
             configured.
 
     Examples:
-        ```pycon
-        >>> source = planetary_variables_source(
+        ```python
+        >>> source = planetary_variable_source(
         ...     "soil_water_content",
         ...     "SWC-AMSR2-C_V1.0_100",
         ...     geometry={

tests/integration/test_subscriptions_api.py

@@ -128,20 +128,25 @@ def result_pages(status=None, size=40):
 # must disable the default.
 res_api_mock = respx.mock(assert_all_called=False)
 
-# 1. Request for status: created. Response has three pages.
+# 1. CSV results
+res_api_mock.route(
+    M(url__startswith=TEST_URL), M(params__contains={'format': 'csv'})).mock(
+        side_effect=[Response(200, text="id,status\n1234-abcd,SUCCESS\n")])
+
+# 2. Request for status: created. Response has three pages.
 res_api_mock.route(
     M(url__startswith=TEST_URL),
     M(params__contains={'status': 'created'})).mock(side_effect=[
         Response(200, json=page)
         for page in result_pages(status={'created'}, size=40)
     ])
 
-# 2. Request for status: queued. Response has a single empty page.
+# 3. Request for status: queued. Response has a single empty page.
 res_api_mock.route(M(url__startswith=TEST_URL),
                    M(params__contains={'status': 'queued'})).mock(
                        side_effect=[Response(200, json={'results': []})])
 
-# 3. No status requested. Response is the same as for 1.
+# 4. No status requested. Response is the same as for 1.
 res_api_mock.route(M(url__startswith=TEST_URL)).mock(
     side_effect=[Response(200, json=page) for page in result_pages(size=40)])
 
@@ -276,6 +281,18 @@ async def test_get_results_success():
         assert len(results) == 100
 
 
+@pytest.mark.anyio
+@res_api_mock
+async def test_get_results_csv():
+    """Subscription CSV fetched, has the expected items."""
+    async with Session() as session:
+        client = SubscriptionsClient(session, base_url=TEST_URL)
+        results = [res async for res in client.get_results_csv("42")]
+        import csv
+        rows = list(csv.reader(results))
+        assert rows == [['id', 'status'], ['1234-abcd', 'SUCCESS']]
+
+
 paging_cycle_api_mock = respx.mock()
 
 # Identical next links is a hangup we want to avoid.

tests/integration/test_subscriptions_cli.py

@@ -306,3 +306,11 @@ def test_request_catalog_success(invoke, geom_geojson):
     ])
     assert json.loads(result.output) == source
     assert result.exit_code == 0  # success.
+
+
+@res_api_mock
+def test_subscriptions_results_csv(invoke):
+    """Get results as CSV."""
+    result = invoke(['results', 'test', '--csv'])
+    assert result.exit_code == 0  # success.
+    assert result.output.splitlines() == ['id,status', '1234-abcd,SUCCESS']

tests/unit/test_subscription_request.py

@@ -349,22 +349,25 @@ def test_toar_tool_success():
     assert res == expected
 
 
-def test_pv_source_success(geom_geojson):
+@pytest.mark.parametrize(
+    "var_type, var_id",
+    [
+        ("biomass_proxy", "BIOMASS-PROXY_V3.0_10"),  # actual real type and id.
+        ("var1", "VAR1-ABCD"),  # nonsense type and id
+    ])
+def test_pv_source_success(geom_geojson, var_type, var_id):
     """Configure a planetary variable subscription source."""
-    # NOTE: this function does not yet validate type and id.
-    # The nonsense values are intended to fail when the function does
-    # add validation.
     source = subscription_request.planetary_variable_source(
-        "var1",
-        "VAR1-abcd",
+        var_type,
+        var_id,
         geometry=geom_geojson,
         start_time=datetime(2021, 3, 1),
         end_time=datetime(2021, 3, 2),
     )
 
-    assert source["type"] == "var1"
+    assert source["type"] == var_type
     params = source["parameters"]
-    assert params["id"] == "VAR1-abcd"
+    assert params["id"] == var_id
     assert params["geometry"] == geom_geojson
     assert params["start_time"].startswith("2021-03-01")