From aea6525c521fe19541af33d8a43b5bb38b63e9db Mon Sep 17 00:00:00 2001 From: asonnenschein Date: Mon, 8 Jun 2026 22:50:55 -0400 Subject: [PATCH 1/3] add poc quota api sdk/cli --- planet/__init__.py | 3 +- planet/cli/cli.py | 3 +- planet/cli/quota.py | 330 +++++++++++++++++++++++++++++ planet/clients/__init__.py | 3 + planet/clients/quota.py | 422 +++++++++++++++++++++++++++++++++++++ planet/sync/client.py | 3 + planet/sync/quota.py | 156 ++++++++++++++ 7 files changed, 918 insertions(+), 2 deletions(-) create mode 100644 planet/cli/quota.py create mode 100644 planet/clients/quota.py create mode 100644 planet/sync/quota.py diff --git a/planet/__init__.py b/planet/__init__.py index 41a9e62b..e80e928e 100644 --- a/planet/__init__.py +++ b/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', diff --git a/planet/cli/cli.py b/planet/cli/cli.py index 467b1e5b..d678ddd0 100644 --- a/planet/cli/cli.py +++ b/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, quota, subscriptions, features LOGGER = logging.getLogger(__name__) @@ -131,6 +131,7 @@ def _configure_logging(verbosity): main.add_command(features.features) # type: ignore main.add_command(destinations.destinations) # type: ignore main.add_command(mosaics.mosaics) # type: ignore +main.add_command(quota.quota) # type: ignore if __name__ == "__main__": main() # pylint: disable=E1120 diff --git a/planet/cli/quota.py b/planet/cli/quota.py new file mode 100644 index 00000000..7c18c080 --- /dev/null +++ b/planet/cli/quota.py @@ -0,0 +1,330 @@ +# Copyright 2026 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. +"""Quota Reservations CLI""" +from contextlib import asynccontextmanager +from typing import Dict, List, Optional, Tuple + +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 +from . import types + + +@asynccontextmanager +async def quota_client(ctx): + async with CliSession(ctx) as sess: + cl = QuotaClient(sess, base_url=ctx.obj['BASE_URL']) + yield cl + + +def _parse_filters(raw: Tuple[str, ...]) -> Optional[Dict[str, str]]: + """Parse repeated `--filter KEY=VALUE` options into a dict.""" + if not raw: + return None + out: Dict[str, str] = {} + for entry in raw: + if '=' not in entry: + raise click.BadParameter( + f"--filter expects KEY=VALUE; got {entry!r}") + key, value = entry.split('=', 1) + key = key.strip() + if not key: + raise click.BadParameter(f"--filter has empty key: {entry!r}") + out[key] = value + return out + + +@click.group() # type: ignore +@click.pass_context +@click.option('-u', + '--base-url', + default=None, + help='Assign custom base Account API URL (e.g. ' + 'https://api.planet.com/account/v1).') +def quota(ctx, base_url): + """Commands for interacting with the Quota Reservations API.""" + ctx.obj['BASE_URL'] = base_url + + +# --------------------------------------------------------------------------- +# Products +# --------------------------------------------------------------------------- + + +@quota.group() +def products(): + """Commands for inspecting products that support quota reservations.""" + pass + + +@command(products, name='list', extra_args=[compact]) +@click.option( + '--supports-reservation/--no-supports-reservation', + default=None, + help='Filter products by whether they support quota reservations.') +async def products_list(ctx, supports_reservation, pretty, compact): + """List products available to your organization. + + Each product's `id` is the value to pass as `--product-id` when creating + or estimating a reservation. + + Example: + + planet quota products list --supports-reservation + """ + async with quota_client(ctx) as cl: + results = await cl.list_products( + supports_reservation=supports_reservation) + if compact: + keys = ('id', + 'name', + 'title', + 'supports_reservation', + 'quota_total', + 'quota_used', + 'unlimited_quota') + results = [{ + k: v + for k, v in p.items() if k in keys + } for p in results] + echo_json(results, pretty) + + +# --------------------------------------------------------------------------- +# Reservations +# --------------------------------------------------------------------------- + + +@quota.group() +def reservations(): + """Commands for managing quota reservations.""" + pass + + +@command(reservations, name='list', extra_args=[limit]) +@click.option('--fields', + default=None, + help='Comma-separated list of fields to include in results.') +@click.option('--sort', + default=None, + help='Sort spec, e.g. `created_at` or `-created_at`.') +@click.option('--filter', + 'filters', + multiple=True, + metavar='KEY=VALUE', + help='Filter by `{field}` or `{field}__{op}`. May be repeated.') +@click.option('--page-size', + type=click.INT, + default=None, + help='Number of reservations to return per page.') +async def reservations_list(ctx, + pretty, + limit, + fields, + sort, + filters, + page_size): + """List quota reservations. + + Example: + + planet quota reservations list --sort -created_at --filter state=active + """ + async with quota_client(ctx) as cl: + results = cl.list_reservations( + limit=limit, + fields=fields, + sort=sort, + filters=_parse_filters(filters), + page_size=page_size, + ) + async for item in results: + echo_json(item, pretty) + + +@command(reservations, name='get') +@click.argument('reservation_id', type=click.INT) +async def reservation_get(ctx, reservation_id, pretty): + """Get a single quota reservation by ID. + + Example: + + planet quota reservations get 100 + """ + async with quota_client(ctx) as cl: + result = await cl.get_reservation(reservation_id) + echo_json(result, pretty) + + +def _aoi_refs_from_options(aoi_ref: Tuple[str, ...], + aoi_refs_json: Optional[List[str]]) -> List[str]: + refs: List[str] = list(aoi_ref) + if aoi_refs_json: + refs.extend(aoi_refs_json) + if not refs: + raise click.BadParameter( + 'At least one AOI ref must be supplied via --aoi-ref or ' + '--aoi-refs.') + return refs + + +_aoi_ref_opt = click.option('--aoi-ref', + multiple=True, + metavar='REF', + help='An AOI feature reference. May be repeated. ' + 'Example: pl:features/my/my-collection/feature-id') + +_aoi_refs_json_opt = click.option( + '--aoi-refs', + 'aoi_refs_json', + type=types.JSON(), + default=None, + help='JSON array of AOI feature refs (string, filename, or `-` for stdin).' +) + +_product_id_opt = click.option('--product-id', + type=click.INT, + required=True, + help='Product ID from `/my/products`.') + +_collection_id_opt = click.option( + '--collection-id', + default=None, + help='Optional grouping collection for the reservation.') + + +@command(reservations, name='create') +@_aoi_ref_opt +@_aoi_refs_json_opt +@_product_id_opt +@_collection_id_opt +async def reservation_create(ctx, + aoi_ref, + aoi_refs_json, + product_id, + collection_id, + pretty): + """Create one or more quota reservations. + + Example: + + \b + planet quota reservations create \\ + --aoi-ref pl:features/my/my-collection/feature-id \\ + --product-id 123 + """ + refs = _aoi_refs_from_options(aoi_ref, aoi_refs_json) + async with quota_client(ctx) as cl: + result = await cl.create_reservation(refs, product_id, collection_id) + echo_json(result, pretty) + + +@command(reservations, name='bulk-reserve') +@_aoi_ref_opt +@_aoi_refs_json_opt +@_product_id_opt +@_collection_id_opt +async def reservation_bulk_reserve(ctx, + aoi_ref, + aoi_refs_json, + product_id, + collection_id, + pretty): + """Submit a bulk quota reservation job (asynchronous). + + Returns a payload with `job_id` and `status`. Track progress with + `planet quota jobs get JOB_ID`. + """ + refs = _aoi_refs_from_options(aoi_ref, aoi_refs_json) + async with quota_client(ctx) as cl: + result = await cl.bulk_create_reservations(refs, + product_id, + collection_id) + echo_json(result, pretty) + + +@command(reservations, name='estimate') +@_aoi_ref_opt +@_aoi_refs_json_opt +@_product_id_opt +@_collection_id_opt +async def reservation_estimate(ctx, + aoi_ref, + aoi_refs_json, + product_id, + collection_id, + pretty): + """Estimate the quota cost of a reservation without creating it.""" + refs = _aoi_refs_from_options(aoi_ref, aoi_refs_json) + async with quota_client(ctx) as cl: + result = await cl.estimate_reservation(refs, product_id, collection_id) + echo_json(result, pretty) + + +# --------------------------------------------------------------------------- +# Jobs +# --------------------------------------------------------------------------- + + +@quota.group() +def jobs(): + """Commands for tracking bulk quota reservation jobs.""" + pass + + +@command(jobs, name='list', extra_args=[limit]) +@click.option('--fields', + default=None, + help='Comma-separated list of fields to include in results.') +@click.option('--sort', default=None, help='Sort spec, e.g. `id` or `-id`.') +@click.option('--filter', + 'filters', + multiple=True, + metavar='KEY=VALUE', + help='Filter by `{field}` or `{field}__{op}`. May be repeated.') +@click.option('--page-size', + type=click.INT, + default=None, + help='Number of jobs to return per page.') +async def jobs_list(ctx, pretty, limit, fields, sort, filters, page_size): + """List bulk quota reservation jobs.""" + async with quota_client(ctx) as cl: + results = cl.list_jobs( + limit=limit, + fields=fields, + sort=sort, + filters=_parse_filters(filters), + page_size=page_size, + ) + async for item in results: + echo_json(item, pretty) + + +@command(jobs, name='get') +@click.argument('job_id') +async def job_get(ctx, job_id, pretty): + """Get the status of a bulk reservation job. + + Example: + + planet quota jobs get 7b7e3a3a-... + """ + async with quota_client(ctx) as cl: + result = await cl.get_job(job_id) + echo_json(result, pretty) diff --git a/planet/clients/__init__.py b/planet/clients/__init__.py index 6aae646f..d831dd1f 100644 --- a/planet/clients/__init__.py +++ b/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 } diff --git a/planet/clients/quota.py b/planet/clients/quota.py new file mode 100644 index 00000000..90b331d8 --- /dev/null +++ b/planet/clients/quota.py @@ -0,0 +1,422 @@ +# Copyright 2026 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. +"""Planet Quota Reservations API Python client.""" + +import logging +from typing import Any, AsyncIterator, Dict, List, Optional, TypeVar + +from planet.clients.base import _BaseClient +from planet.exceptions import APIError, ClientError +from planet.http import Session +from planet.models import Paged, Response +from ..constants import PLANET_BASE_URL + +BASE_URL = f'{PLANET_BASE_URL}/account/v1' + +LOGGER = logging.getLogger() + +T = TypeVar("T") + + +class _QuotaPaged(Paged): + """Pager for Quota API list responses. + + Quota API list responses have the shape: + {"meta": {"count": N, "next": "", "prev": ""}, "results": [...]} + """ + ITEMS_KEY = 'results' + + def _next_link(self, page): + try: + next_link = page['meta']['next'] + except KeyError: + next_link = False + if not next_link: + LOGGER.debug('end of the pages') + return next_link or False + + +class QuotaClient(_BaseClient): + """Asynchronous Quota Reservations API client. + + The methods of this class forward request parameters to the operations + described in the Planet Quota Reservations API + (https://api.planet.com/account/v1/quota-reservations/spec). + + For more information, see the documentation at + https://docs.planet.com/develop/apis/quota/ + + 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) -> None: + """ + Parameters: + session: Open session connected to server. + base_url: The base URL to use. Defaults to the production Account + API base URL at api.planet.com (`/account/v1`). + """ + super().__init__(session, base_url or BASE_URL) + self._reservations_url = f'{self._base_url}/quota-reservations' + self._products_url = f'{self._base_url}/my/products' + + @staticmethod + def _filter_params( + fields: Optional[str] = None, + sort: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Build a query-string params dict shared by list endpoints. + + The Quota API accepts `fields`, `sort`, and arbitrary `{field}__{op}` / + `{field}` filter pairs as query parameters. + """ + params: Dict[str, Any] = {} + if fields is not None: + params['fields'] = fields + if sort is not None: + params['sort'] = sort + if filters: + params.update(filters) + return params + + async def list_reservations( + self, + limit: int = 100, + fields: Optional[str] = None, + sort: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[dict]: + """Iterate over quota reservations. + + Parameters: + limit: Maximum number of reservations to return. When set to 0, + no maximum is applied. + fields: Comma-separated list of model fields to return. + sort: Sort spec - `` for ascending or `-` for + descending. + filters: Mapping of arbitrary filter parameters. Keys may be + `{field}` (equality) or `{field}__{op}` where `op` is one of + `eq`, `lt`, `lte`, `gt`, `gte`, `ne`, `like`, `ilike`, + `icontains`, `isnull`, `in`, `notin`. + page_size: Number of results to fetch per page. + + Yields: + dict: A description of a quota reservation. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + params = self._filter_params(fields=fields, sort=sort, filters=filters) + if page_size is not None: + params['limit'] = page_size + + url = f'{self._reservations_url}/' + try: + response = await self._session.request(method='GET', + url=url, + params=params) + async for item in _QuotaPaged(response, + self._session.request, + limit=limit): + yield item + except APIError: + raise + except ClientError: # pragma: no cover + raise + + async def get_reservation(self, reservation_id: int) -> dict: + """Get a single quota reservation by ID. + + Parameters: + reservation_id: ID of the quota reservation. + + Returns: + dict: description of the quota reservation. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + url = f'{self._reservations_url}/{reservation_id}' + try: + resp = await self._session.request(method='GET', url=url) + except APIError: + raise + except ClientError: # pragma: no cover + raise + return resp.json() + + async def create_reservation( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> dict: + """Create one or more quota reservations. + + Parameters: + aoi_refs: List of AOI feature references (e.g. + `pl:features/my/collection-id/feature-id`). Supplying a + collection ref will reserve every feature in that collection. + product_id: The product ID from `/my/products`. + collection_id: Optional grouping collection for the reservation. + + Returns: + dict: a `QuotaReservationCreated` payload including + `quota_total`, `quota_used`, `quota_remaining`, and a list of + created reservation refs. + + Raises: + APIError: on an API server error (e.g. insufficient quota). + ClientError: on a client error. + """ + body: Dict[str, Any] = { + 'aoi_refs': list(aoi_refs), + 'product_id': product_id, + } + if collection_id is not None: + body['collection_id'] = collection_id + + url = f'{self._reservations_url}/' + try: + resp = await self._session.request(method='POST', + url=url, + json=body) + except APIError: + raise + except ClientError: # pragma: no cover + raise + return resp.json() + + async def bulk_create_reservations( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> dict: + """Create quota reservations asynchronously via a bulk job. + + Use this endpoint for large batches of AOI references. The response + includes a `job_id` whose progress can be tracked with + [`get_job`][planet.clients.quota.QuotaClient.get_job]. + + Parameters: + aoi_refs: List of AOI feature references. + product_id: The product ID from `/my/products`. + collection_id: Optional grouping collection for the reservation. + + Returns: + dict: payload with `job_id` and `status`. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + body: Dict[str, Any] = { + 'aoi_refs': list(aoi_refs), + 'product_id': product_id, + } + if collection_id is not None: + body['collection_id'] = collection_id + + url = f'{self._reservations_url}/bulk-reserve' + try: + resp = await self._session.request(method='POST', + url=url, + json=body) + except APIError: + raise + except ClientError: # pragma: no cover + raise + return resp.json() + + async def estimate_reservation( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> dict: + """Estimate the quota cost of a reservation without creating it. + + To verify there is sufficient quota for the reservation, compare the + returned `total_cost` against `quota_remaining` - the remaining value + does not pre-subtract the estimate. + + Parameters: + aoi_refs: List of AOI feature references. + product_id: The product ID from `/my/products`. + collection_id: Optional grouping collection for the reservation. + + Returns: + dict: a `QuotaReservationEstimation` payload including + `total_cost`, `estimated_costs`, `quota_total`, + `quota_remaining`, and `quota_units`. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + body: Dict[str, Any] = { + 'aoi_refs': list(aoi_refs), + 'product_id': product_id, + } + if collection_id is not None: + body['collection_id'] = collection_id + + url = f'{self._reservations_url}/estimate' + try: + resp = await self._session.request(method='POST', + url=url, + json=body) + except APIError: + raise + except ClientError: # pragma: no cover + raise + return resp.json() + + async def list_jobs( + self, + limit: int = 100, + fields: Optional[str] = None, + sort: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[dict]: + """Iterate over bulk quota reservation jobs. + + Parameters: + limit: Maximum number of jobs to return. When set to 0, no + maximum is applied. + fields: Comma-separated list of model fields to return. + sort: Sort spec - `` for ascending or `-` for + descending. + filters: Mapping of `{field}` / `{field}__{op}` filter pairs. + page_size: Number of results to fetch per page. + + Yields: + dict: A description of a quota reservation job. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + params = self._filter_params(fields=fields, sort=sort, filters=filters) + if page_size is not None: + params['limit'] = page_size + + url = f'{self._reservations_url}/jobs' + try: + response = await self._session.request(method='GET', + url=url, + params=params) + async for item in _QuotaPaged(response, + self._session.request, + limit=limit): + yield item + except APIError: + raise + except ClientError: # pragma: no cover + raise + + async def get_job(self, job_id: str) -> dict: + """Get the status of a bulk reservation job. + + Parameters: + job_id: ID of the bulk reservation job. + + Returns: + dict: description of the job, including `status`, + `processed_items`, `total_items`, and `percentage`. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + if not job_id: + raise ClientError("Must provide a job id") + + url = f'{self._reservations_url}/jobs/{job_id}' + try: + resp = await self._session.request(method='GET', url=url) + except APIError: + raise + except ClientError: # pragma: no cover + raise + return resp.json() + + async def list_products( + self, + supports_reservation: Optional[bool] = None, + ) -> List[dict]: + """List products available to the requesting user's organization. + + Use this to look up the `product_id` (the `id` field) to pass into + [`create_reservation`][planet.clients.quota.QuotaClient.create_reservation], + [`bulk_create_reservations`][planet.clients.quota.QuotaClient.bulk_create_reservations], + or [`estimate_reservation`][planet.clients.quota.QuotaClient.estimate_reservation]. + + Parameters: + supports_reservation: If True, only return products with + `supports_reservation: true`. If False, only return products + without reservation support. If None (default), return all + accessible products. + + Returns: + list[dict]: products in the user's organization. + + Raises: + APIError: on an API server error. + ClientError: on a client error. + """ + try: + resp: Response = await self._session.request( + method='GET', url=self._products_url) + except APIError: + raise + except ClientError: # pragma: no cover + raise + + payload = resp.json() + # /my/products returns either a list or a dict wrapping a list under + # a 'results' / 'products' key; tolerate both shapes. + if isinstance(payload, dict): + products = payload.get('results') or payload.get('products') or [] + else: + products = payload + + if supports_reservation is None: + return list(products) + return [ + p for p in products + if bool(p.get('supports_reservation')) == supports_reservation + ] + + +__all__ = ['QuotaClient'] diff --git a/planet/sync/client.py b/planet/sync/client.py index 993b3527..499e6fe6 100644 --- a/planet/sync/client.py +++ b/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__ @@ -22,6 +23,7 @@ class Planet: - `data`: for interacting with the Planet Data API. - `destinations`: Destinations API. - `orders`: Orders API. + - `quota`: Quota Reservations API. - `subscriptions`: Subscriptions API. - `features`: Features API @@ -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}/account/v1") diff --git a/planet/sync/quota.py b/planet/sync/quota.py new file mode 100644 index 00000000..dea57b00 --- /dev/null +++ b/planet/sync/quota.py @@ -0,0 +1,156 @@ +# Copyright 2026 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. +"""Synchronous Planet Quota Reservations API client.""" + +from typing import Any, Dict, Iterator, List, Optional + +from planet.clients.quota import QuotaClient +from planet.http import Session + + +class QuotaAPI: + """Quota Reservations API client. + + Example: + ```python + >>> from planet import Planet + >>> + >>> pl = Planet() + >>> for r in pl.quota.list_reservations(): + ... print(r) + ``` + """ + + _client: QuotaClient + + def __init__(self, + session: Session, + base_url: Optional[str] = None) -> None: + """ + Parameters: + session: Open session connected to server. + base_url: The base URL to use. Defaults to the production Account + API base URL. + """ + self._client = QuotaClient(session, base_url) + + def list_reservations( + self, + limit: int = 100, + fields: Optional[str] = None, + sort: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + page_size: Optional[int] = None, + ) -> Iterator[dict]: + """Iterate over quota reservations. + + See [QuotaClient.list_reservations][planet.clients.quota.QuotaClient.list_reservations] + for parameter details. + """ + return self._client._aiter_to_iter( + self._client.list_reservations(limit=limit, + fields=fields, + sort=sort, + filters=filters, + page_size=page_size)) + + def get_reservation(self, reservation_id: int) -> Dict[str, Any]: + """Get a single quota reservation by ID.""" + return self._client._call_sync( + self._client.get_reservation(reservation_id)) + + def create_reservation( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Create one or more quota reservations. + + Parameters: + aoi_refs: List of AOI feature references (e.g. + `pl:features/my/collection-id/feature-id`). + product_id: The product ID from `/my/products`. + collection_id: Optional grouping collection for the reservation. + """ + return self._client._call_sync( + self._client.create_reservation(aoi_refs, + product_id, + collection_id)) + + def bulk_create_reservations( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Submit a bulk quota reservation job. + + Returns a payload with `job_id` and `status` - track progress with + [get_job][planet.sync.quota.QuotaAPI.get_job]. + """ + return self._client._call_sync( + self._client.bulk_create_reservations(aoi_refs, + product_id, + collection_id)) + + def estimate_reservation( + self, + aoi_refs: List[str], + product_id: int, + collection_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Estimate the quota cost of a reservation without creating it.""" + return self._client._call_sync( + self._client.estimate_reservation(aoi_refs, + product_id, + collection_id)) + + def list_jobs( + self, + limit: int = 100, + fields: Optional[str] = None, + sort: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + page_size: Optional[int] = None, + ) -> Iterator[dict]: + """Iterate over bulk quota reservation jobs.""" + return self._client._aiter_to_iter( + self._client.list_jobs(limit=limit, + fields=fields, + sort=sort, + filters=filters, + page_size=page_size)) + + def get_job(self, job_id: str) -> Dict[str, Any]: + """Get the status of a bulk reservation job.""" + return self._client._call_sync(self._client.get_job(job_id)) + + def list_products( + self, + supports_reservation: Optional[bool] = None, + ) -> List[Dict[str, Any]]: + """List products available to the requesting user's organization. + + Use this to look up the `product_id` (the `id` field) to pass into + [create_reservation][planet.sync.quota.QuotaAPI.create_reservation], + [bulk_create_reservations][planet.sync.quota.QuotaAPI.bulk_create_reservations], + or [estimate_reservation][planet.sync.quota.QuotaAPI.estimate_reservation]. + + Parameters: + supports_reservation: If True, only return products with + `supports_reservation: true`. + """ + return self._client._call_sync( + self._client.list_products(supports_reservation)) From 7845c2b68b465ca8f534c6daccb78871ca94d92c Mon Sep 17 00:00:00 2001 From: asonnenschein Date: Tue, 9 Jun 2026 20:20:25 -0400 Subject: [PATCH 2/3] add tests --- tests/integration/test_quota_api.py | 380 ++++++++++++++++++++++++++++ tests/integration/test_quota_cli.py | 358 ++++++++++++++++++++++++++ 2 files changed, 738 insertions(+) create mode 100644 tests/integration/test_quota_api.py create mode 100644 tests/integration/test_quota_cli.py diff --git a/tests/integration/test_quota_api.py b/tests/integration/test_quota_api.py new file mode 100644 index 00000000..b29a4351 --- /dev/null +++ b/tests/integration/test_quota_api.py @@ -0,0 +1,380 @@ +# Copyright 2026 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. +"""Tests of the Planet Quota Reservations API client.""" +from http import HTTPStatus +import json +from typing import Any, Optional + +import httpx +import pytest +import respx + +from planet import QuotaClient, Session +from planet.auth import Auth +from planet.exceptions import APIError, ClientError +from planet.sync.quota import QuotaAPI + +pytestmark = pytest.mark.anyio # noqa + +# Simulated host/path for testing purposes. Not a real subdomain. +TEST_URL = "http://test.planet.com/account/v1" +RESERVATIONS_URL = f"{TEST_URL}/quota-reservations/" +JOBS_URL = f"{TEST_URL}/quota-reservations/jobs" +PRODUCTS_URL = f"{TEST_URL}/my/products" + +AOI_REF = "pl:features/my/test-collection/feature-1" + +# Set up shared test clients (mirrors test_features_api.py). +test_session = Session(auth=Auth.from_key(key="test")) +cl_async = QuotaClient(test_session, base_url=TEST_URL) +cl_sync = QuotaAPI(test_session, base_url=TEST_URL) + + +def mock_response(url: str, + json: Any, + method: str = "get", + status_code: int = HTTPStatus.OK): + """Register a single canned response on the respx router.""" + respx.request(method, url).return_value = httpx.Response(status_code, + json=json) + + +def _reservation(rid: int) -> dict: + return { + "id": rid, + "aoi_ref": AOI_REF, + "product_id": 100, + "state": "active", + "created_at": "2026-01-01T00:00:00Z", + } + + +def _reservations_page(start: int, + end: int, + next_url: Optional[str] = None) -> dict: + page: dict = { + "meta": { + "count": end - start + }, + "results": [_reservation(i) for i in range(start, end)], + } + if next_url is not None: + page["meta"]["next"] = next_url + return page + + +def _job(jid: str = "job-abc") -> dict: + return { + "id": jid, + "status": "complete", + "processed_items": 2, + "total_items": 2, + "percentage": 100, + } + + +# --------------------------------------------------------------------------- +# list_reservations +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_list_reservations_single_page(): + mock_response(RESERVATIONS_URL, _reservations_page(0, 3)) + + def assertf(resp): + assert [r["id"] for r in resp] == [0, 1, 2] + + assertf([r async for r in cl_async.list_reservations()]) + assertf(list(cl_sync.list_reservations())) + + +@respx.mock +async def test_list_reservations_paginated(): + """Follow the meta.next link until exhausted.""" + next_url = f"{RESERVATIONS_URL}?cursor=abc" + # First page links to next_url, second page (matched via params) has no next. + respx.get(RESERVATIONS_URL).mock(side_effect=[ + httpx.Response(200, json=_reservations_page(0, 2, next_url=next_url)), + httpx.Response(200, json=_reservations_page(2, 4)) + ]) + + items = [r async for r in cl_async.list_reservations()] + assert [r["id"] for r in items] == [0, 1, 2, 3] + + +@respx.mock +async def test_list_reservations_respects_limit(): + # Two pages of two items each — limit=3 cuts iteration short. + next_url = f"{RESERVATIONS_URL}?cursor=abc" + respx.get(RESERVATIONS_URL).mock(side_effect=[ + httpx.Response(200, json=_reservations_page(0, 2, next_url=next_url)), + httpx.Response(200, json=_reservations_page(2, 4)) + ]) + + items = [r async for r in cl_async.list_reservations(limit=3)] + assert [r["id"] for r in items] == [0, 1, 2] + + +@respx.mock +async def test_list_reservations_query_params(): + mock_response(RESERVATIONS_URL, _reservations_page(0, 1)) + + _ = [ + r async for r in cl_async.list_reservations( + fields="id,state", + sort="-created_at", + filters={ + "state": "active", "product_id__in": "1,2" + }, + page_size=25, + ) + ] + + sent = respx.calls[0].request.url.params + assert sent["fields"] == "id,state" + assert sent["sort"] == "-created_at" + assert sent["state"] == "active" + assert sent["product_id__in"] == "1,2" + assert sent["limit"] == "25" + + +# --------------------------------------------------------------------------- +# get_reservation +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_get_reservation(): + rid = 42 + mock_response(f"{TEST_URL}/quota-reservations/{rid}", _reservation(rid)) + + def assertf(resp): + assert resp["id"] == rid + + assertf(await cl_async.get_reservation(rid)) + assertf(cl_sync.get_reservation(rid)) + + +@respx.mock +async def test_get_reservation_api_error(): + rid = 99 + mock_response(f"{TEST_URL}/quota-reservations/{rid}", + json={"message": "not found"}, + status_code=HTTPStatus.NOT_FOUND) + with pytest.raises(APIError): + await cl_async.get_reservation(rid) + + +# --------------------------------------------------------------------------- +# create_reservation +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_create_reservation(): + payload = { + "quota_total": 1000, + "quota_used": 10, + "quota_remaining": 990, + "reservation_refs": ["pl:reservations/1"], + } + mock_response(RESERVATIONS_URL, payload, method="post") + + def assertf(resp): + assert resp == payload + + assertf(await cl_async.create_reservation([AOI_REF], 100)) + assertf(cl_sync.create_reservation([AOI_REF], 100)) + + req_body = json.loads(respx.calls[0].request.content) + assert req_body == {"aoi_refs": [AOI_REF], "product_id": 100} + + +@respx.mock +async def test_create_reservation_with_collection_id(): + mock_response(RESERVATIONS_URL, {}, method="post") + await cl_async.create_reservation([AOI_REF], 100, collection_id="col-1") + req_body = json.loads(respx.calls[0].request.content) + assert req_body["collection_id"] == "col-1" + + +@respx.mock +async def test_create_reservation_omits_collection_id_when_none(): + mock_response(RESERVATIONS_URL, {}, method="post") + await cl_async.create_reservation([AOI_REF], 100) + req_body = json.loads(respx.calls[0].request.content) + assert "collection_id" not in req_body + + +# --------------------------------------------------------------------------- +# bulk_create_reservations +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_bulk_create_reservations(): + bulk_url = f"{TEST_URL}/quota-reservations/bulk-reserve" + payload = {"job_id": "job-abc", "status": "queued"} + mock_response(bulk_url, payload, method="post") + + def assertf(resp): + assert resp == payload + + assertf(await cl_async.bulk_create_reservations([AOI_REF], 100)) + assertf(cl_sync.bulk_create_reservations([AOI_REF], 100)) + + req_body = json.loads(respx.calls[0].request.content) + assert req_body == {"aoi_refs": [AOI_REF], "product_id": 100} + + +# --------------------------------------------------------------------------- +# estimate_reservation +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_estimate_reservation(): + estimate_url = f"{TEST_URL}/quota-reservations/estimate" + payload = { + "total_cost": 5, + "estimated_costs": [{ + "aoi_ref": AOI_REF, "cost": 5 + }], + "quota_total": 100, + "quota_remaining": 90, + "quota_units": "sqkm", + } + mock_response(estimate_url, payload, method="post") + + def assertf(resp): + assert resp == payload + + assertf(await cl_async.estimate_reservation([AOI_REF], 100, "col-1")) + assertf(cl_sync.estimate_reservation([AOI_REF], 100, "col-1")) + + req_body = json.loads(respx.calls[0].request.content) + assert req_body == { + "aoi_refs": [AOI_REF], + "product_id": 100, + "collection_id": "col-1", + } + + +# --------------------------------------------------------------------------- +# Jobs +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_list_jobs(): + page = {"meta": {"count": 2}, "results": [_job("a"), _job("b")]} + mock_response(JOBS_URL, page) + + def assertf(resp): + assert [j["id"] for j in resp] == ["a", "b"] + + assertf([j async for j in cl_async.list_jobs()]) + assertf(list(cl_sync.list_jobs())) + + +@respx.mock +async def test_get_job(): + job_id = "job-xyz" + mock_response(f"{JOBS_URL}/{job_id}", _job(job_id)) + + def assertf(resp): + assert resp["id"] == job_id + + assertf(await cl_async.get_job(job_id)) + assertf(cl_sync.get_job(job_id)) + + +async def test_get_job_empty_id_raises(): + with pytest.raises(ClientError): + await cl_async.get_job("") + + +# --------------------------------------------------------------------------- +# Products +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_list_products_list_payload(): + products = [ + { + "id": 1, "supports_reservation": True + }, + { + "id": 2, "supports_reservation": False + }, + ] + mock_response(PRODUCTS_URL, products) + + def assertf(resp): + assert [p["id"] for p in resp] == [1, 2] + + assertf(await cl_async.list_products()) + assertf(cl_sync.list_products()) + + +@respx.mock +async def test_list_products_results_wrapper(): + """`/my/products` may wrap items in `results` — accept that shape too.""" + payload = { + "results": [{ + "id": 7, "supports_reservation": True + }], + } + mock_response(PRODUCTS_URL, payload) + products = await cl_async.list_products() + assert [p["id"] for p in products] == [7] + + +@respx.mock +async def test_list_products_products_wrapper(): + """And the `products` wrapper key is also tolerated.""" + payload = { + "products": [{ + "id": 8, "supports_reservation": False + }], + } + mock_response(PRODUCTS_URL, payload) + products = await cl_async.list_products() + assert [p["id"] for p in products] == [8] + + +@respx.mock +async def test_list_products_supports_reservation_filter(): + products = [ + { + "id": 1, "supports_reservation": True + }, + { + "id": 2, "supports_reservation": False + }, + { + "id": 3, "supports_reservation": True + }, + ] + # Each filter invocation makes a fresh request — return the same list each time. + respx.get(PRODUCTS_URL).mock( + side_effect=lambda req: httpx.Response(200, json=products)) + + supported = await cl_async.list_products(supports_reservation=True) + assert [p["id"] for p in supported] == [1, 3] + + unsupported = await cl_async.list_products(supports_reservation=False) + assert [p["id"] for p in unsupported] == [2] diff --git a/tests/integration/test_quota_cli.py b/tests/integration/test_quota_cli.py new file mode 100644 index 00000000..66fb8fd1 --- /dev/null +++ b/tests/integration/test_quota_cli.py @@ -0,0 +1,358 @@ +# Copyright 2026 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. +"""Tests of the planet quota CLI.""" +import json +import tempfile + +import respx +from click.testing import CliRunner + +from planet.cli import cli + +from tests.integration.test_quota_api import ( + AOI_REF, + JOBS_URL, + PRODUCTS_URL, + RESERVATIONS_URL, + TEST_URL, + _job, + _reservation, + _reservations_page, + mock_response, +) + + +def invoke(*args, input=None): + runner = CliRunner() + full_args = ["quota", "--base-url", TEST_URL] + list(args) + result = runner.invoke(cli.main, args=full_args, input=input) + assert result.exit_code == 0, result.output + return result + + +def _parse_json_lines(output: str): + """`echo_json` prints one JSON document per line — parse them all.""" + return [json.loads(line) for line in output.splitlines() if line.strip()] + + +# --------------------------------------------------------------------------- +# products +# --------------------------------------------------------------------------- + + +@respx.mock +def test_cli_products_list(): + products = [ + { + "id": 1, + "name": "PSScene", + "title": "PlanetScope Scene", + "supports_reservation": True, + "quota_total": 10, + "quota_used": 1, + "unlimited_quota": False, + "extra": "dropped-when-compact", + }, + { + "id": 2, + "name": "OtherProduct", + "supports_reservation": False, + "quota_total": 0, + "quota_used": 0, + "unlimited_quota": True, + "extra": "still-here-without-compact", + }, + ] + mock_response(PRODUCTS_URL, products) + + # Default: every key surfaces. + result = invoke("products", "list") + data = json.loads(result.output) + assert [p["id"] for p in data] == [1, 2] + assert data[0]["extra"] == "dropped-when-compact" + + +@respx.mock +def test_cli_products_list_compact(): + products = [ + { + "id": 1, + "name": "PSScene", + "title": "PlanetScope Scene", + "supports_reservation": True, + "quota_total": 10, + "quota_used": 1, + "unlimited_quota": False, + "extra": "dropped-when-compact", + }, + ] + mock_response(PRODUCTS_URL, products) + + result = invoke("products", "list", "--compact") + data = json.loads(result.output) + assert "extra" not in data[0] + assert set(data[0].keys()) == { + "id", + "name", + "title", + "supports_reservation", + "quota_total", + "quota_used", + "unlimited_quota", + } + + +@respx.mock +def test_cli_products_list_supports_reservation_flag(): + products = [ + { + "id": 1, "supports_reservation": True + }, + { + "id": 2, "supports_reservation": False + }, + ] + respx.get(PRODUCTS_URL).respond(json=products) + + result = invoke("products", "list", "--supports-reservation") + assert [p["id"] for p in json.loads(result.output)] == [1] + + +# --------------------------------------------------------------------------- +# reservations list / get +# --------------------------------------------------------------------------- + + +@respx.mock +def test_cli_reservations_list(): + mock_response(RESERVATIONS_URL, _reservations_page(0, 3)) + result = invoke("reservations", "list") + items = _parse_json_lines(result.output) + assert [i["id"] for i in items] == [0, 1, 2] + + +@respx.mock +def test_cli_reservations_list_passes_filters(): + mock_response(RESERVATIONS_URL, _reservations_page(0, 1)) + invoke( + "reservations", + "list", + "--limit", + "5", + "--sort", + "-created_at", + "--fields", + "id,state", + "--filter", + "state=active", + "--filter", + "product_id__in=1,2", + "--page-size", + "50", + ) + params = respx.calls[0].request.url.params + assert params["sort"] == "-created_at" + assert params["fields"] == "id,state" + assert params["state"] == "active" + assert params["product_id__in"] == "1,2" + assert params["limit"] == "50" + + +def test_cli_reservations_list_bad_filter(): + runner = CliRunner() + result = runner.invoke( + cli.main, + args=[ + "quota", + "--base-url", + TEST_URL, + "reservations", + "list", + "--filter", + "no_equals_sign", + ], + ) + assert result.exit_code != 0 + assert "--filter" in result.output + + +@respx.mock +def test_cli_reservations_get(): + rid = 42 + mock_response(f"{TEST_URL}/quota-reservations/{rid}", _reservation(rid)) + result = invoke("reservations", "get", str(rid)) + assert json.loads(result.output)["id"] == rid + + +# --------------------------------------------------------------------------- +# reservations create / bulk-reserve / estimate +# --------------------------------------------------------------------------- + + +@respx.mock +def test_cli_reservation_create_aoi_ref_flags(): + """Repeated --aoi-ref flags accumulate into a list.""" + payload = {"reservation_refs": ["pl:reservations/1"]} + mock_response(RESERVATIONS_URL, payload, method="post") + + result = invoke( + "reservations", + "create", + "--aoi-ref", + AOI_REF, + "--aoi-ref", + "pl:features/my/c/f2", + "--product-id", + "100", + ) + assert json.loads(result.output) == payload + + body = json.loads(respx.calls[0].request.content) + assert body == { + "aoi_refs": [AOI_REF, "pl:features/my/c/f2"], + "product_id": 100, + } + + +@respx.mock +def test_cli_reservation_create_aoi_refs_json_string(): + """`--aoi-refs '[...]'` passes a JSON array directly.""" + mock_response(RESERVATIONS_URL, {}, method="post") + invoke( + "reservations", + "create", + "--aoi-refs", + json.dumps([AOI_REF]), + "--product-id", + "100", + "--collection-id", + "col-1", + ) + + body = json.loads(respx.calls[0].request.content) + assert body == { + "aoi_refs": [AOI_REF], + "product_id": 100, + "collection_id": "col-1", + } + + +@respx.mock +def test_cli_reservation_create_aoi_refs_from_file(): + mock_response(RESERVATIONS_URL, {}, method="post") + with tempfile.NamedTemporaryFile("w+", suffix=".json") as f: + json.dump([AOI_REF, "pl:features/my/c/f2"], f) + f.flush() + invoke( + "reservations", + "create", + "--aoi-refs", + f.name, + "--product-id", + "100", + ) + + body = json.loads(respx.calls[0].request.content) + assert body["aoi_refs"] == [AOI_REF, "pl:features/my/c/f2"] + + +@respx.mock +def test_cli_reservation_create_aoi_refs_from_stdin(): + mock_response(RESERVATIONS_URL, {}, method="post") + invoke( + "reservations", + "create", + "--aoi-refs", + "-", + "--product-id", + "100", + input=json.dumps([AOI_REF]), + ) + body = json.loads(respx.calls[0].request.content) + assert body["aoi_refs"] == [AOI_REF] + + +def test_cli_reservation_create_requires_aoi_refs(): + """Without --aoi-ref or --aoi-refs the command fails before any HTTP call.""" + runner = CliRunner() + result = runner.invoke( + cli.main, + args=[ + "quota", + "--base-url", + TEST_URL, + "reservations", + "create", + "--product-id", + "100", + ], + ) + assert result.exit_code != 0 + assert "AOI ref" in result.output + + +@respx.mock +def test_cli_reservation_bulk_reserve(): + bulk_url = f"{TEST_URL}/quota-reservations/bulk-reserve" + payload = {"job_id": "job-abc", "status": "queued"} + mock_response(bulk_url, payload, method="post") + + result = invoke( + "reservations", + "bulk-reserve", + "--aoi-ref", + AOI_REF, + "--product-id", + "100", + ) + assert json.loads(result.output) == payload + + +@respx.mock +def test_cli_reservation_estimate(): + estimate_url = f"{TEST_URL}/quota-reservations/estimate" + payload = {"total_cost": 5, "quota_remaining": 95} + mock_response(estimate_url, payload, method="post") + + result = invoke( + "reservations", + "estimate", + "--aoi-ref", + AOI_REF, + "--product-id", + "100", + ) + assert json.loads(result.output) == payload + + +# --------------------------------------------------------------------------- +# jobs list / get +# --------------------------------------------------------------------------- + + +@respx.mock +def test_cli_jobs_list(): + page = {"meta": {"count": 2}, "results": [_job("a"), _job("b")]} + mock_response(JOBS_URL, page) + result = invoke("jobs", "list") + items = _parse_json_lines(result.output) + assert [j["id"] for j in items] == ["a", "b"] + + +@respx.mock +def test_cli_jobs_get(): + job_id = "job-xyz" + mock_response(f"{JOBS_URL}/{job_id}", _job(job_id)) + result = invoke("jobs", "get", job_id) + assert json.loads(result.output)["id"] == job_id From 6f037697dbf0388631fb220b3e7a996bc11d421e Mon Sep 17 00:00:00 2001 From: asonnenschein Date: Tue, 9 Jun 2026 22:15:09 -0400 Subject: [PATCH 3/3] fix linter issues --- planet/cli/types.py | 6 ++++-- tests/integration/test_quota_api.py | 3 +-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/planet/cli/types.py b/planet/cli/types.py index c3168ea5..d768c490 100644 --- a/planet/cli/types.py +++ b/planet/cli/types.py @@ -24,7 +24,8 @@ class CommaSeparatedString(click.types.StringParamType): """A list of strings that is extracted from a comma-separated string.""" - def convert(self, value, param, ctx) -> List[str]: + def convert( # type: ignore[override] + self, value, param, ctx) -> List[str]: if isinstance(value, list): convlist = value else: @@ -46,7 +47,8 @@ class CommaSeparatedFloat(click.types.StringParamType): """A list of floats that is extracted from a comma-separated string.""" name = 'VALUE' - def convert(self, value, param, ctx) -> List[float]: + def convert( # type: ignore[override] + self, value, param, ctx) -> List[float]: values = CommaSeparatedString().convert(value, param, ctx) try: diff --git a/tests/integration/test_quota_api.py b/tests/integration/test_quota_api.py index b29a4351..6be21f3d 100644 --- a/tests/integration/test_quota_api.py +++ b/tests/integration/test_quota_api.py @@ -138,8 +138,7 @@ async def test_list_reservations_query_params(): filters={ "state": "active", "product_id__in": "1,2" }, - page_size=25, - ) + page_size=25, ) ] sent = respx.calls[0].request.url.params