Quota API

planet/__init__.py

@@ -17,7 +17,7 @@
 from .__version__ import __version__  # NOQA
 from .auth import Auth
 from .auth_builtins import PlanetOAuthScopes
-from .clients import DataClient, DestinationsClient, FeaturesClient, MosaicsClient, OrdersClient, SubscriptionsClient  # NOQA
+from .clients import DataClient, DestinationsClient, FeaturesClient, MosaicsClient, OrdersClient, QuotaClient, SubscriptionsClient  # NOQA
 from .io import collect
 from .sync import Planet
 
@@ -33,6 +33,7 @@
     'OrdersClient',
     'order_request',
     'Planet',
+    'QuotaClient',
     'reporting',
     'Session',
     'SubscriptionsClient',

planet/cli/cli.py

@@ -22,7 +22,7 @@
 import planet
 from planet.cli import mosaics
 
-from . import auth, cmds, collect, data, destinations, orders, subscriptions, features
+from . import auth, cmds, collect, data, destinations, orders, subscriptions, features, quota
 
 LOGGER = logging.getLogger(__name__)
 
@@ -130,6 +130,7 @@ def _configure_logging(verbosity):
 main.add_command(collect.collect)  # type: ignore
 main.add_command(features.features)  # type: ignore
 main.add_command(destinations.destinations)  # type: ignore
+main.add_command(quota.quota)  # type: ignore
 main.add_command(mosaics.mosaics)  # type: ignore
 
 if __name__ == "__main__":

planet/cli/quota.py

@@ -0,0 +1,135 @@
+# Copyright 2025 Planet Labs PBC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+from contextlib import asynccontextmanager
+import json
+import click
+from planet.cli.io import echo_json
+from planet.clients.quota import QuotaClient
+from .cmds import command
+from .options import compact, limit
+from .session import CliSession
+
+
+@asynccontextmanager
+async def quota_client(ctx):
+    async with CliSession() as sess:
+        cl = QuotaClient(sess, base_url=ctx.obj['BASE_URL'])
+        yield cl
+
+
+@click.group()  # type: ignore
+@click.pass_context
+@click.option('-u',
+              '--base-url',
+              default=None,
+              help='Assign custom base Quota API URL.')
+def quota(ctx, base_url):
+    """Commands for interacting with the Quota API"""
+    ctx.obj['BASE_URL'] = base_url
+
+
+@quota.group()  # type: ignore
+def reservations():
+    """Commands for managing quota reservations"""
+    pass
+
+
+@command(reservations, name="create")
+@click.argument("request_file", type=click.Path(exists=True))
+async def reservation_create(ctx, request_file, pretty):
+    """Create a new quota reservation from a JSON request file.
+    Example:
+    \b
+    planet quota reservations create ./reservation_request.json
+    """
+    async with quota_client(ctx) as cl:
+        with open(request_file) as f:
+            request = json.load(f)
+        result = await cl.create_reservation(request)
+        echo_json(result, pretty)
+
+
+@command(reservations, name="list", extra_args=[limit, compact])
+@click.option("--status", help="Filter reservations by status")
+async def reservations_list(ctx, pretty, limit, compact, status):
+    """List quota reservations.
+    Example:
+    \b
+    planet quota reservations list
+    planet quota reservations list --status active
+    """
+    async with quota_client(ctx) as cl:
+        results = cl.list_reservations(status=status, limit=limit)
+        if compact:
+            compact_fields = ('id', 'name', 'status', 'created_at')
+            output = [{
+                k: v
+                for k, v in row.items() if k in compact_fields
+            } async for row in results]
+        else:
+            output = [r async for r in results]
+        echo_json(output, pretty)
+
+
+@command(reservations, name="get")
+@click.argument("reservation_id", required=True)
+async def reservation_get(ctx, reservation_id, pretty):
+    """Get a quota reservation by ID.
+    Example:
+    \b
+    planet quota reservations get 12345678-1234-5678-9012-123456789012
+    """
+    async with quota_client(ctx) as cl:
+        result = await cl.get_reservation(reservation_id)
+        echo_json(result, pretty)
+
+
+@command(reservations, name="cancel")
+@click.argument("reservation_id", required=True)
+async def reservation_cancel(ctx, reservation_id, pretty):
+    """Cancel an existing quota reservation.
+    Example:
+    \b
+    planet quota reservations cancel 12345678-1234-5678-9012-123456789012
+    """
+    async with quota_client(ctx) as cl:
+        result = await cl.cancel_reservation(reservation_id)
+        echo_json(result, pretty)
+
+
+@command(quota, name="estimate")
+@click.argument("request_file", type=click.Path(exists=True))
+async def quota_estimate(ctx, request_file, pretty):
+    """Estimate quota requirements for a potential reservation.
+    Example:
+    \b
+    planet quota estimate ./estimation_request.json
+    """
+    async with quota_client(ctx) as cl:
+        with open(request_file) as f:
+            request = json.load(f)
+        result = await cl.estimate_quota(request)
+        echo_json(result, pretty)
+
+
+@command(quota, name="usage")
+async def quota_usage(ctx, pretty):
+    """Get current quota usage and limits.
+    Example:
+    \b
+    planet quota usage
+    """
+    async with quota_client(ctx) as cl:
+        result = await cl.get_quota_usage()
+        echo_json(result, pretty)

planet/clients/__init__.py

@@ -17,6 +17,7 @@
 from .features import FeaturesClient
 from .mosaics import MosaicsClient
 from .orders import OrdersClient
+from .quota import QuotaClient
 from .subscriptions import SubscriptionsClient
 
 __all__ = [
@@ -25,6 +26,7 @@
     'FeaturesClient',
     'MosaicsClient',
     'OrdersClient',
+    'QuotaClient',
     'SubscriptionsClient'
 ]
 
@@ -35,5 +37,6 @@
     'features': FeaturesClient,
     'mosaics': MosaicsClient,
     'orders': OrdersClient,
+    'quota': QuotaClient,
     'subscriptions': SubscriptionsClient
 }

planet/clients/quota.py

@@ -0,0 +1,153 @@
+# Copyright 2025 Planet Labs PBC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+import logging
+from typing import AsyncIterator, Optional
+from planet.clients.base import _BaseClient
+from planet.constants import PLANET_BASE_URL
+from planet.http import Session
+from planet.models import Paged
+
+BASE_URL = f'{PLANET_BASE_URL}/quota/v1'
+LOGGER = logging.getLogger(__name__)
+
+
+class Reservations(Paged):
+    """Asynchronous iterator over reservations from a paged response."""
+    NEXT_KEY = '_next'
+    ITEMS_KEY = 'reservations'
+
+
+class QuotaClient(_BaseClient):
+    """High-level asynchronous access to Planet's quota API.
+    The Planet Quota Reservations API allows you to create, estimate, and view
+    existing quota reservations on the Planet platform for compatible products
+    including Planetary Variables, Analysis-Ready PlanetScope (ARPS), and select
+    PlanetScope imagery products.
+    Example:
+        ```python
+        >>> import asyncio
+        >>> from planet import Session
+        >>>
+        >>> async def main():
+        ...     async with Session() as sess:
+        ...         cl = sess.client('quota')
+        ...         # use client here
+        ...
+        >>> asyncio.run(main())
+        ```
+    """
+
+    def __init__(self, session: Session, base_url: Optional[str] = None):
+        """
+        Parameters:
+            session: Open session connected to server.
+            base_url: The base URL to use. Defaults to production quota API
+                base url.
+        """
+        super().__init__(session, base_url or BASE_URL)
+
+    def _reservations_url(self):
+        return f'{self._base_url}/reservations'
+
+    async def create_reservation(self, request: dict) -> dict:
+        """Create a new quota reservation.
+        Parameters:
+            request: Quota reservation request specification.
+        Returns:
+            Description of the created reservation.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = self._reservations_url()
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
+        return response.json()
+
+    async def estimate_quota(self, request: dict) -> dict:
+        """Estimate quota requirements for a potential reservation.
+        Parameters:
+            request: Quota estimation request specification.
+        Returns:
+            Quota estimation details including projected costs and usage.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = f'{self._base_url}/estimate'
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
+        return response.json()
+
+    async def list_reservations(self,
+                                status: Optional[str] = None,
+                                limit: int = 100) -> AsyncIterator[dict]:
+        """Iterate through list of quota reservations.
+        Parameters:
+            status: Filter reservations by status (e.g., 'active', 'completed', 'cancelled').
+            limit: Maximum number of results to return. When set to 0, no
+                maximum is applied.
+        Yields:
+            Description of a quota reservation.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = self._reservations_url()
+        params = {}
+        if status:
+            params['status'] = status
+        response = await self._session.request(method='GET',
+                                               url=url,
+                                               params=params)
+        async for reservation in Reservations(response,
+                                              self._session.request,
+                                              limit=limit):
+            yield reservation
+
+    async def get_reservation(self, reservation_id: str) -> dict:
+        """Get a quota reservation by ID.
+        Parameters:
+            reservation_id: Quota reservation identifier.
+        Returns:
+            Quota reservation details.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = f'{self._reservations_url()}/{reservation_id}'
+        response = await self._session.request(method='GET', url=url)
+        return response.json()
+
+    async def cancel_reservation(self, reservation_id: str) -> dict:
+        """Cancel an existing quota reservation.
+        Parameters:
+            reservation_id: Quota reservation identifier.
+        Returns:
+            Updated reservation details.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = f'{self._reservations_url()}/{reservation_id}/cancel'
+        response = await self._session.request(method='POST', url=url)
+        return response.json()
+
+    async def get_quota_usage(self) -> dict:
+        """Get current quota usage and limits.
+        Returns:
+            Current quota usage statistics and limits.
+        Raises:
+            planet.exceptions.APIError: On API error.
+        """
+        url = f'{self._base_url}/usage'
+        response = await self._session.request(method='GET', url=url)
+        return response.json()

planet/sync/client.py

@@ -4,6 +4,7 @@
 from .data import DataAPI
 from .destinations import DestinationsAPI
 from .orders import OrdersAPI
+from .quota import QuotaAPI
 from .subscriptions import SubscriptionsAPI
 from planet.http import Session
 from planet.__version__ import __version__
@@ -24,6 +25,7 @@ class Planet:
     - `orders`: Orders API.
     - `subscriptions`: Subscriptions API.
     - `features`: Features API
+    - `quota`: Quota API
 
     Quick start example:
     ```python
@@ -66,3 +68,4 @@ def __init__(self,
             self._session, f"{planet_base}/subscriptions/v1/")
         self.features = FeaturesAPI(self._session,
                                     f"{planet_base}/features/v1/ogc/my/")
+        self.quota = QuotaAPI(self._session, f"{planet_base}/quota/v1")