Repository: planetlabs/planet-client-python
Pull request #555 (2022-05-18 02:18:10)

implement search-quick functionality


Patch:
diff --git a/planet/cli/data.py b/planet/cli/data.py
index d655681..a663dad 100644
--- a/planet/cli/data.py
+++ b/planet/cli/data.py
@@ -1,17 +1,38 @@
 """The Planet Data CLI."""
 
 import json
-from typing import AsyncIterator, List
+from typing import List
+from contextlib import asynccontextmanager
 
 import click
+import planet
+from planet import DataClient, Session
 
 from .cmds import coro, translate_exceptions
 from .io import echo_json
 
-# Parameter callbacks are what we use to validate and transform the
-# values of arguments and options. On the command line, everything is
-# a string, but we want to transform to natural Python types as soon as
-# possible.
+pretty = click.option('--pretty', is_flag=True, help='Pretty-print output.')
+
+
+@asynccontextmanager
+async def data_client(ctx):
+    auth = ctx.obj['AUTH']
+    base_url = ctx.obj['BASE_URL']
+    async with Session(auth=auth) as sess:
+        cl = DataClient(sess, base_url=base_url)
+        yield cl
+
+
+@click.group()
+@click.pass_context
+@click.option('-u',
+              '--base-url',
+              default=None,
+              help='Assign custom base Orders API URL.')
+def data(ctx, base_url):
+    '''Commands for interacting with the Orders API'''
+    ctx.obj['AUTH'] = planet.Auth.from_file()
+    ctx.obj['BASE_URL'] = base_url
 
 
 def parse_item_types(ctx, param, value: str) -> List[str]:
@@ -23,83 +44,69 @@ def parse_item_types(ctx, param, value: str) -> List[str]:
 
 def parse_filter(ctx, param, value: str) -> dict:
     """Turn filter JSON into a dict."""
-    return json.loads(value)
-
-
-@click.group()
-@click.pass_context
-def data(ctx):
-    """A group of commands for interacting with the Data API."""
+    # read filter using raw json
+    if value.startswith('{'):
+        try:
+            json_value = json.loads(value)
+        except json.decoder.JSONDecodeError:
+            raise click.BadParameter('Filter does not contain valid json.',
+                                     ctx=ctx,
+                                     param=param)
+        if json_value == {}:
+            raise click.BadParameter('Filter is empty.', ctx=ctx, param=param)
+        return json_value
+    # read filter using click pipe option
+    else:
+        try:
+            with click.open_file(value) as f:
+                json_value = json.load(f)
+        except json.decoder.JSONDecodeError:
+            raise click.BadParameter('Filter does not contain valid json.',
+                                     ctx=ctx,
+                                     param=param)
+        return json_value
 
 
 # TODO: filter().
 
 
 @data.command()
+@click.pass_context
 @translate_exceptions
-# Our CLI command functions are async functions. They need an event
-# loop to be executed. The coro decorator provides that event loop.
 @coro
-# The command has two positional arguments. ITEM_TYPES comes first.
-# By default these are required and will evaluate to str type.
+@pretty
 @click.argument("item_types", callback=parse_item_types)
 @click.argument("filter", callback=parse_filter)
-# The command has three options.
-@click.option("--limit",
+@click.option('--name',
+              type=str,
+              default=False,
+              help=('Name of the saved search.'))
+@click.option('--limit',
               type=int,
               default=100,
-              help="Maximum number of results to return.")
-@click.option("--name", help="Name for the saved search.")
-@click.option("--pretty", is_flag=True, help="Pretty print output.")
-# This decorator gives our function a context (named "ctx") that may
-# contain parameters set by the "planet data" command group, like
-# "planet --quiet data" or "planet --verbosity=debug data".
-@click.pass_context
-async def search_quick(ctx,
-                       item_types: List[str],
-                       filter: dict,
-                       limit: int,
-                       pretty: bool,
-                       name: str):
-    """Execute a structured item search and print results.
+              help='Maximum number of results to return. Defaults to 100.')
+async def search_quick(ctx, item_types, filter, name, limit, pretty):
+    """Execute a structured item search.
+    Quick searches are stored for approximately 30 days and the --name
+    parameter will be applied to the stored quick search. \n
 
-    Results are represented as a sequence of GeoJSON features.
+    Arguments: \n
+    ITEM_TYPES - string. Comma-separated item type identifier(s). \n
+    FILTER - string. A full JSON description of search criteria.
+    Supports file and stdin. \n
 
-    """
+    Output:
+    A series of GeoJSON descriptions for each of the returned items.
 
-    # Note: API tokens will be found in "ctx". They are unused in this
-    # example.
-    #
-    # The Planet Data client will return a stream of GeoJSON
-    # Feature-like dicts (aka an "iterator"). The search_quick function
-    # will iterate over the features, encode them as JSON, and echo
-    # them. We simulate that here.
-    async def example_client(item_types,
-                             filter,
-                             limit=0,
-                             name=None) -> AsyncIterator[dict]:
-        """This is a placeholder.
-
-        We'll delete this and use the real method from
-        planet.clients.data when it is available.
-
-        """
-        # Note: we re-emit the input item types and filter to help us
-        # sanity check during early sprints.
-        for feature in item_types:
-            yield dict(item_type=feature)
-
-        yield dict(filter=filter)
-
-    # CLI functions raise ClickException when any kind of intentionally
-    # raised PlanetError occurs. We *don't* handle other errors now
-    # because those will be caused by bugs in our code and we want them
-    # raw and unfiltered.
-    async for feature in example_client(item_types,
-                                        filter,
-                                        limit=limit,
-                                        name=name):
-        echo_json(feature, pretty=pretty)
+    """
+    async with data_client(ctx) as cl:
+        items = await cl.quick_search(name=name,
+                                      item_types=item_types,
+                                      search_filter=filter,
+                                      limit=limit,
+                                      sort=None)
+        async for item in items:
+            echo_json(item, pretty)
 
 
 # TODO: search_create()".


Test patch:
diff --git a/tests/integration/test_data_cli.py b/tests/integration/test_data_cli.py
index 8f62a79..9a0b460 100644
--- a/tests/integration/test_data_cli.py
+++ b/tests/integration/test_data_cli.py
@@ -1,20 +1,84 @@
 """Tests of the Data CLI."""
-
 from click.testing import CliRunner
 import pytest
-
+import json
+import httpx
+import respx
+from http import HTTPStatus
 from planet.cli import cli
 
+TEST_URL = 'https://api.planet.com/data/v1'
+TEST_QUICKSEARCH_URL = f'{TEST_URL}/quick-search'
+
+
+@pytest.fixture
+def invoke():
 
-def test_data_command_registered():
+    def _invoke(extra_args, runner=None):
+        runner = runner or CliRunner()
+        args = ['data'] + extra_args
+        return runner.invoke(cli.main, args=args)
+
+    return _invoke
+
+
+def test_data_command_registered(invoke):
     """planet-data command prints help and usage message."""
-    result = CliRunner().invoke(cli.main, ["data", "--help"])
+    runner = CliRunner()
+    result = invoke(["--help"], runner=runner)
     assert result.exit_code == 0
     assert "Usage" in result.output
     assert "search-quick" in result.output
     # Add other sub-commands here.
 
 
+@respx.mock
+@pytest.mark.asyncio
+@pytest.mark.parametrize("filter", ['{1:1}', '{"foo"}'])
+@pytest.mark.parametrize(
+    "item_types", ['PSScene', 'SkySatScene', ('PSScene', 'SkySatScene')])
+def test_data_search_quick_filter_fail(invoke, item_types, filter):
+    """Test for planet data search_quick. Test with multiple item_types.
+    Test should fail as filter does not contain valid JSON."""
+    mock_resp = httpx.Response(HTTPStatus.OK,
+                               json={'features': [{
+                                   "key": "value"
+                               }]})
+    respx.post(TEST_QUICKSEARCH_URL).return_value = mock_resp
+
+    runner = CliRunner()
+    result = invoke(["search-quick", item_types, filter], runner=runner)
+    assert result.exit_code == 2
+
+
+@respx.mock
+@pytest.mark.parametrize(
+    "item_types", ['PSScene', 'SkySatScene', ('PSScene', 'SkySatScene')])
+def test_data_search_quick_filter_success(invoke, item_types):
+    """Test for planet data search_quick. Test with multiple item_types.
+    Test should succeed as filter contains valid JSON."""
+    filter = {
+        "type": "DateRangeFilter",
+        "field_name": "acquired",
+        "config": {
+            "gt": "2019-12-31T00:00:00Z", "lte": "2020-01-31T00:00:00Z"
+        }
+    }
+
+    mock_resp = httpx.Response(HTTPStatus.OK,
+                               json={'features': [{
+                                   "key": "value"
+                               }]})
+    respx.post(TEST_QUICKSEARCH_URL).return_value = mock_resp
+
+    runner = CliRunner()
+    result = invoke(["search-quick", item_types, json.dumps(filter)],
+                    runner=runner)
+
+    assert result.exit_code == 0
+    assert len(result.output.strip().split('\n')) == 1  # we have 1 feature
+
+
 # TODO: basic test for "planet data filter".