import itertools as it
from typing import Any, Dict, Iterator, List, Optional, cast
from requests.exceptions import HTTPError
from ..exceptions import EntityDoesNotExist, MLHubException
from ..session import get_session
[docs]def list_collections(*, api_key: Optional[str] = None, profile: Optional[str] = None) -> List[Dict[str, Any]]:
"""Gets a list of JSON-like dictionaries representing STAC Collection objects returned by the Radiant MLHub ``GET /collections``
endpoint.
See the `MLHub API docs <https://docs.mlhub.earth/#radiant-mlhub-api>`_ for details.
Parameters
----------
api_key : str
An API key to use for this request. This will override an API key set in a profile on using
an environment variable
profile: str
A profile to use when making this request.
Returns
-------
collections: List[dict]
List of JSON-like dictionaries representing STAC Collection objects.
"""
session = get_session(api_key=api_key, profile=profile)
r = session.get('collections')
return cast(List[Dict[str, Any]], r.json().get('collections', []))
[docs]def get_collection(collection_id: str, *, api_key: Optional[str] = None, profile: Optional[str] = None) -> Dict[str, Any]:
"""Returns a JSON-like dictionary representing the response from the Radiant MLHub ``GET /collections/{p1}`` endpoint.
See the `MLHub API docs <https://docs.mlhub.earth/#radiant-mlhub-api>`_ for details.
Parameters
----------
collection_id : str
The ID of the collection to fetch
api_key : str
An API key to use for this request. This will override an API key set in a profile on using
an environment variable
profile: str
A profile to use when making this request.
Returns
-------
collection : dict
Raises
------
EntityDoesNotExist
If a 404 response code is returned by the API
MLHubException
If any other response code is returned
"""
session = get_session(api_key=api_key, profile=profile)
try:
return cast(Dict[str, Any], session.get(f'collections/{collection_id}').json())
except HTTPError as e:
if e.response.status_code == 404:
raise EntityDoesNotExist(f'Collection "{collection_id}" does not exist.') from None
raise MLHubException(f'An unknown error occurred: {e.response.status_code} ({e.response.reason})') from None
[docs]def list_collection_items(
collection_id: str,
*,
page_size: Optional[int] = None,
extensions: Optional[List[str]] = None,
limit: int = 10,
api_key: Optional[str] = None,
profile: Optional[str] = None
) -> Iterator[Dict[str, Any]]:
"""Yields JSON-like dictionaries representing STAC Item objects returned by the Radiant MLHub ``GET /collections/{collection_id}/items``
endpoint.
.. note::
Because some collections may contain hundreds of thousands of items, this function limits the total number of responses
to ``10`` by default. You can change this value by increasing the value of the ``limit`` keyword argument,
or setting it to ``None`` to list all items. **Be aware that trying to list all items in a large collection may take a very
long time.**
Parameters
----------
collection_id : str
The ID of the collection from which to fetch items
page_size : int
The number of items to return in each page. If set to ``None``, then this parameter will not be passed to the API and
the default API value will be used (currently ``30``).
extensions : list
If provided, then only items that support all of the extensions listed will be returned.
limit : int
The maximum *total* number of items to yield. Defaults to ``10``.
api_key : str
An API key to use for this request. This will override an API key set in a profile on using
an environment variable
profile: str
A profile to use when making this request.
Yields
------
item : dict
JSON-like dictionary representing a STAC Item associated with the given collection.
"""
session = get_session(api_key=api_key, profile=profile)
def _list_items() -> Iterator[Dict[str, Any]]:
params: Dict[str, Any] = {}
if page_size is not None:
params['limit'] = page_size
if extensions is not None:
params['extensions'] = extensions
for page in session.paginate(f'collections/{collection_id}/items', params=params):
yield from page['features']
yield from it.islice(_list_items(), limit)
[docs]def get_collection_item(collection_id: str, item_id: str, api_key: Optional[str] = None, profile: Optional[str] = None) -> Dict[str, Any]:
"""Returns a JSON-like dictionary representing the response from the Radiant MLHub ``GET /collections/{p1}/items/{p2}`` endpoint.
Parameters
----------
collection_id : str
The ID of the Collection to which the Item belongs.
item_id : str
The ID of the Item.
api_key : str
An API key to use for this request. This will override an API key set in a profile on using
an environment variable
profile: str
A profile to use when making this request.
Returns
-------
item : dict
"""
session = get_session(api_key=api_key, profile=profile)
try:
return cast(Dict[str, Any], session.get(f'collections/{collection_id}/items/{item_id}').json())
except HTTPError as e:
if e.response.status_code == 404:
raise EntityDoesNotExist(f'Collection "{collection_id}" and/or item {item_id} do not exist.')
raise MLHubException(f'An unknown error occurred: {e.response.status_code} ({e.response.reason})')