Support bulk subs creation in client and add --bulk to CLI to invoke it

docs/cli/cli-subscriptions.md

@@ -247,6 +247,49 @@ planet subscriptions patch cb817760-1f07-4ee7-bba6-bcac5346343f \
     patched-attributes.json
 ```
 
+### Bulk Create Subscriptions
+
+To create many subscriptions for different geometries at once, use the `bulk-create` subcommand.
+
+This command allows submitting a bulk create request that references a feature collection defined in
+the Features API, which will create a subscription for every feature in the collection.
+
+Define a subscription that references a feature collection:
+
+```json
+{
+  "name": "new guinea psscene bulk subscription",
+  "source": {
+    "parameters": {
+      "item_types": [
+        "PSScene"
+      ],
+      "asset_types": [
+        "ortho_visual"
+      ],
+      "geometry": {
+        "type": "ref",
+        "content": "pl:features/my/test-new-guinea-10geojson-xqRXaaZ"
+      },
+      "start_time": "2021-01-01T00:00:00Z",
+      "end_time": "2021-01-05T00:00:00Z"
+    }
+  }
+}
+```
+
+And issue the `bulk-create` command with the appropriate hosting or delivery options. A link to list
+the resulting subscriptions will be returned:
+
+```sh
+planet subscriptions bulk-create --hosting sentinel_hub catalog_fc_sub.json
+{
+  "_links": {
+    "list": "https://api.planet.com/subscriptions/v1?created=2025-04-16T23%3A44%3A35Z%2F..&geom_ref=pl%3Afeatures%2Fmy%2Ftest-new-guinea-10geojson-xqRXaaZ&name=new+guinea+psscene+bulk subscription"
+  }
+}
+```
+
 ### Cancel Subscription
 
 Cancelling a subscription is simple with the CLI:

planet/cli/subscriptions.py

@@ -203,6 +203,60 @@ async def create_subscription_cmd(ctx, request, pretty, **kwargs):
         echo_json(sub, pretty)
 
 
+@subscriptions.command(name="bulk-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 hosting. '
+              'If omitted, a new collection will be created.')
+@click.option(
+    '--create-configuration',
+    is_flag=True,
+    help='Automatically create a layer configuration for your collection. '
+    'If omitted, no configuration will be created.')
+@pretty
+@click.pass_context
+@translate_exceptions
+@coro
+async def bulk_create_subscription_cmd(ctx, request, pretty, **kwargs):
+    """Bulk create subscriptions.
+
+    Submits a bulk subscription request for creation and prints a link to list
+    the resulting subscriptions.
+
+    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, collection_id, and create_configuration.
+    The hosting flag specifies the hosting type, the collection_id flag specifies the
+    collection ID for Sentinel Hub, and the create_configuration flag specifies
+    whether or not to create a layer configuration for your collection. If the
+    collection_id is omitted, a new collection will be created. If the
+    create_configuration flag is omitted, no configuration will be created. The
+    collection_id flag and create_configuration flag cannot be used together.
+    """
+    hosting = kwargs.get("hosting", None)
+    collection_id = kwargs.get("collection_id", None)
+    create_configuration = kwargs.get('create_configuration', False)
+
+    if hosting == "sentinel_hub":
+        hosting_info = sentinel_hub(collection_id, create_configuration)
+        request["hosting"] = hosting_info
+
+    async with subscriptions_client(ctx) as client:
+        links = await client.bulk_create_subscriptions([request])
+        # Bulk create returns just a link to an endpoint to list created subscriptions.
+        echo_json(links, pretty)
+
+
 @subscriptions.command(name='cancel')  # type: ignore
 @click.argument('subscription_id')
 @pretty

planet/clients/subscriptions.py

@@ -1,7 +1,7 @@
 """Planet Subscriptions API Python client."""
 
 import logging
-from typing import Any, AsyncIterator, Awaitable, Dict, Optional, Sequence, TypeVar
+from typing import Any, AsyncIterator, Awaitable, Dict, Optional, Sequence, TypeVar, List
 
 from typing_extensions import Literal
 
@@ -203,6 +203,34 @@ async def create_subscription(self, request: dict) -> dict:
             sub = resp.json()
             return sub
 
+    async def bulk_create_subscriptions(self, requests: List[dict]) -> Dict:
+        """
+        Create multiple subscriptions in bulk. Currently, the list of requests can only contain one item.
+
+        Args:
+            requests (List[dict]): A list of dictionaries where each dictionary
+            represents a subscription to be created.
+
+        Raises:
+            APIError: If the API returns an error response.
+            ClientError: If there is an issue with the client request.
+
+        Returns:
+            The response including a _links key to the list endpoint for use finding the created subscriptions.
+        """
+        try:
+            url = f'{self._base_url}/bulk'
+            resp = await self._session.request(
+                method='POST', url=url, json={'subscriptions': requests})
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
+        else:
+            return resp.json()
+
     async def cancel_subscription(self, subscription_id: str) -> None:
         """Cancel a Subscription.
 

planet/sync/subscriptions.py

@@ -1,6 +1,6 @@
 """Planet Subscriptions API Python client."""
 
-from typing import Any, Dict, Iterator, Optional, Sequence, Union
+from typing import Any, Dict, Iterator, Optional, Sequence, Union, List
 
 from typing_extensions import Literal
 
@@ -136,6 +136,22 @@ def create_subscription(self, request: Dict) -> Dict:
         return self._client._call_sync(
             self._client.create_subscription(request))
 
+    def bulk_create_subscriptions(self, requests: List[Dict]) -> Dict:
+        """Bulk create subscriptions.
+
+        Args:
+            request (List[dict]): list of descriptions of a bulk creation.
+
+        Returns:
+            response including link to list of created subscriptions
+
+        Raises:
+            APIError: on an API server error.
+            ClientError: on a client error.
+        """
+        return self._client._call_sync(
+            self._client.bulk_create_subscriptions(requests))
+
     def cancel_subscription(self, subscription_id: str) -> None:
         """Cancel a Subscription.
 

tests/integration/test_subscriptions_api.py

@@ -122,6 +122,17 @@ def modify_response(request):
 create_mock.route(M(url=TEST_URL),
                   method='POST').mock(side_effect=modify_response)
 
+bulk_create_mock = respx.mock()
+bulk_create_mock.route(
+    M(url=f'{TEST_URL}/bulk'), method='POST'
+).mock(return_value=Response(
+    200,
+    json={
+        '_links': {
+            'list': f'{TEST_URL}/subscriptions/v1?created={datetime.now().isoformat()}/&geom_ref=pl:features:test_features&name=test-sub'
+        }
+    }))
+
 update_mock = respx.mock()
 update_mock.route(M(url=f'{TEST_URL}/test'),
                   method='PUT').mock(side_effect=modify_response)
@@ -334,6 +345,18 @@ async def test_create_subscription_success():
         assert sub['name'] == 'test'
 
 
+@pytest.mark.anyio
+@bulk_create_mock
+async def test_bulk_create_subscription_success():
+    """Bulk subscription is created, description has the expected items."""
+    async with Session() as session:
+        client = SubscriptionsClient(session, base_url=TEST_URL)
+        resp = await client.bulk_create_subscriptions([{
+            'name': 'test', 'delivery': 'yes, please', 'source': 'test'
+        }])
+        assert '/subscriptions/v1?' in resp['_links']['list']
+
+
 @create_mock
 def test_create_subscription_success_sync():
     """Subscription is created, description has the expected items."""
@@ -346,6 +369,18 @@ def test_create_subscription_success_sync():
     assert sub['name'] == 'test'
 
 
+@bulk_create_mock
+def test_bulk_create_subscription_success_sync():
+    """Subscription is created, description has the expected items."""
+
+    pl = Planet()
+    pl.subscriptions._client._base_url = TEST_URL
+    resp = pl.subscriptions.bulk_create_subscriptions([{
+        'name': 'test', 'delivery': 'yes, please', 'source': 'test'
+    }])
+    assert '/subscriptions/v1?' in resp['_links']['list']
+
+
 @pytest.mark.anyio
 @create_mock
 async def test_create_subscription_with_hosting_success():

tests/integration/test_subscriptions_cli.py

@@ -25,6 +25,7 @@
 from test_subscriptions_api import (api_mock,
                                     cancel_mock,
                                     create_mock,
+                                    bulk_create_mock,
                                     failing_api_mock,
                                     get_mock,
                                     patch_mock,
@@ -138,6 +139,28 @@ def test_subscriptions_create_success(invoke, cmd_arg, runner_input):
     assert result.exit_code == 0  # success.
 
 
+@pytest.mark.parametrize('cmd_arg, runner_input',
+                         [('-', json.dumps(GOOD_SUB_REQUEST)),
+                          (json.dumps(GOOD_SUB_REQUEST), None),
+                          ('-', json.dumps(GOOD_SUB_REQUEST_WITH_HOSTING)),
+                          (json.dumps(GOOD_SUB_REQUEST_WITH_HOSTING), None)])
+@bulk_create_mock
+def test_subscriptions_bulk_create_success(invoke, cmd_arg, runner_input):
+    """Subscriptions creation succeeds with a valid subscription request."""
+
+    # The "-" argument says "read from stdin" and the input keyword
+    # argument specifies what bytes go to the runner's stdin.
+    result = invoke(
+        ['bulk-create', cmd_arg],
+        input=runner_input,
+        # Note: catch_exceptions=True (the default) is required if we want
+        # to exercise the "translate_exceptions" decorator and test for
+        # failure.
+        catch_exceptions=True)
+
+    assert result.exit_code == 0  # success.
+
+
 # Invalid JSON.
 BAD_SUB_REQUEST = '{0: "lolwut"}'