Add Hosting Block Convenience Method to SDK

CHANGES.txt

@@ -1,4 +1,7 @@
+2.5.0 (2024-04-04)
 
+Added:
+- Support Subscriptions API hosting block for Sentinel Hub in the CLI (#1029).
 
 Added:
  - Support for field_boundaries_sentinel_2_p1m in Subscriptions API (#1026)

planet/cli/subscriptions.py

@@ -11,6 +11,7 @@
 from .session import CliSession
 from planet.clients.subscriptions import SubscriptionsClient
 from .. import subscription_request
+from ..subscription_request import sentinel_hub
 from ..specs import get_item_types, validate_item_type, SpecificationException
 
 ALL_ITEM_TYPES = get_item_types()
@@ -87,21 +88,46 @@ async def list_subscriptions_cmd(ctx, status, limit, pretty):
             echo_json(sub, pretty)
 
 
-@subscriptions.command(name='create')  # type: ignore
-@click.argument('request', type=types.JSON())
+@subscriptions.command(name="create")  # type: ignore
+@click.argument("request", type=types.JSON())
+@click.option(
+    "--hosting",
+    type=click.Choice([
+        "sentinel_hub",
+    ]),
+    default=None,
+    help='Hosting type. Currently, only "sentinel_hub" is supported.',
+)
+@click.option("--collection-id",
+              default=None,
+              help='Collection ID for Sentinel Hub.'
+              'If omitted, a new collection will be created.')
 @pretty
 @click.pass_context
 @translate_exceptions
 @coro
-async def create_subscription_cmd(ctx, request, pretty):
+async def create_subscription_cmd(ctx, request, pretty, **kwargs):
     """Create a subscription.
 
     Submits a subscription request for creation and prints the created
     subscription description, optionally pretty-printed.
 
     REQUEST is the full description of the subscription to be created. It must
     be JSON and can be specified a json string, filename, or '-' for stdin.
+
+    Other flag options are hosting and collection_id. The hosting flag
+    specifies the hosting type, and the collection_id flag specifies the
+    collection ID for Sentinel Hub. If the collection_id is omitted, a new
+    collection will be created.
     """
+
+    hosting = kwargs.get("hosting", None)
+    collection_id = kwargs.get("collection_id", None)
+
+    if hosting == "sentinel_hub":
+        hosting_info = sentinel_hub(collection_id)
+        request["hosting"] = hosting_info
+
     async with subscriptions_client(ctx) as client:
         sub = await client.create_subscription(request)
         echo_json(sub, pretty)
@@ -263,20 +289,28 @@ async def list_subscription_results_cmd(ctx,
     help='Toolchain JSON. Can be a string, filename, or - for stdin.')
 @click.option(
     '--hosting',
-    type=types.JSON(),
-    help='Hosting JSON.  Can be a string, a filename, or - for stdin.')
+    default=None,
+    type=click.Choice([
+        "sentinel_hub",
+    ]),
+    help='Hosting configuration. Can be JSON, "sentinel_hub", or omitted.')
 @click.option(
     '--clip-to-source',
     is_flag=True,
     default=False,
     help="Clip to the source geometry without specifying a clip tool.")
+@click.option("--collection-id",
+              default=None,
+              help='Collection ID for Sentinel Hub.'
+              'If omitted, a new collection will be created.')
 @pretty
 def request(name,
             source,
             delivery,
             notifications,
             tools,
             hosting,
+            collection_id,
             clip_to_source,
             pretty):
     """Generate a subscriptions request.
@@ -286,12 +320,14 @@ def request(name,
     --clip-to-source option is a preview of the next API version's
     default behavior.
     """
+
     res = subscription_request.build_request(name,
                                              source,
                                              delivery,
                                              notifications=notifications,
                                              tools=tools,
                                              hosting=hosting,
+                                             collection_id=collection_id,
                                              clip_to_source=clip_to_source)
     echo_json(res, pretty)
 
@@ -349,6 +385,7 @@ def request_catalog(item_types,
                     time_range_type,
                     pretty):
     """Generate a subscriptions request catalog source description."""
+
     res = subscription_request.catalog_source(
         item_types,
         asset_types,

planet/subscription_request.py

@@ -52,7 +52,8 @@ def build_request(name: str,
                   delivery: Optional[Mapping] = None,
                   notifications: Optional[Mapping] = None,
                   tools: Optional[List[Mapping]] = None,
-                  hosting: Optional[Mapping] = None,
+                  hosting: Optional[Union[Mapping, str]] = None,
+                  collection_id: Optional[str] = None,
                   clip_to_source: Optional[bool] = False) -> dict:
     """Construct a Subscriptions API request.
 
@@ -150,8 +151,15 @@ def build_request(name: str,
 
         details['tools'] = tool_list
 
-    if hosting:
-        details['hosting'] = dict(hosting)
+    if hosting == "sentinel_hub":
+        hosting_info: Dict[str, Any] = {
+            "type": "sentinel_hub", "parameters": {}
+        }
+        if collection_id:
+            hosting_info["parameters"]["collection_id"] = collection_id
+        details['hosting'] = hosting_info
+    elif isinstance(hosting, dict):
+        details['hosting'] = hosting
 
     return details
 

tests/integration/test_subscriptions_cli.py

@@ -435,3 +435,44 @@ def test_catalog_source_time_range_type(invoke, geom_geojson, time_range_type):
     assert result.exit_code == 0  # success.
     req = json.loads(result.output)
     assert req['parameters']['time_range_type'] == time_range_type
+
+
+@pytest.mark.parametrize(
+    "hosting_option, collection_id_option, expected_success",
+    [
+        ("--hosting=sentinel_hub", None, True),
+        ("--hosting=sentinel_hub",
+         "--collection-id=7ff105c4-e0de-4910-96db-8345d86ab734",
+         True),
+    ])
+def test_request_hosting(invoke,
+                         geom_geojson,
+                         hosting_option,
+                         collection_id_option,
+                         expected_success):
+    """Test request command with various hosting and collection ID options."""
+    source = json.dumps({
+        "type": "catalog",
+        "parameters": {
+            "geometry": geom_geojson,
+            "start_time": "2021-03-01T00:00:00Z",
+            "end_time": "2023-11-01T00:00:00Z",
+            "rrule": "FREQ=MONTHLY;BYMONTH=3,4,5,6,7,8,9,10",
+            "item_types": ["PSScene"],
+            "asset_types": ["ortho_analytic_4b"]
+        }
+    })
+
+    cmd = [
+        'request',
+        '--name=test',
+        f'--source={source}',
+        hosting_option,
+    ]
+
+    if collection_id_option:
+        cmd.append(collection_id_option)
+
+    result = invoke(cmd)
+
+    assert result.exit_code == 0, "Expected command to succeed."