subscriptions.py

"""Planet Subscriptions API Python client."""

from typing import Any, Dict, Iterator, Optional, Sequence, Union, List

from typing_extensions import Literal

from planet.http import Session
from planet.clients import SubscriptionsClient


class SubscriptionsAPI:
    """Subscriptions API client

    Example:
        ```python
        >>> from planet import Planet
        >>>
        >>> pl = Planet()
        >>> pl.subscriptions.list_subscriptions()
        ```
    """

    _client: SubscriptionsClient

    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 production subscriptions
                API base url.
        """

        self._client = SubscriptionsClient(session, base_url)

    def list_subscriptions(self,
                           status: Optional[Sequence[str]] = None,
                           limit: int = 100,
                           created: Optional[str] = None,
                           end_time: Optional[str] = None,
                           hosting: Optional[bool] = None,
                           name__contains: Optional[str] = None,
                           name: Optional[str] = None,
                           source_type: Optional[str] = None,
                           start_time: Optional[str] = None,
                           sort_by: Optional[str] = None,
                           updated: Optional[str] = None,
                           destination_ref: Optional[str] = None,
                           user_id: Optional[Union[str, int]] = None,
                           geom_ref: Optional[str] = None,
                           page_size: int = 500) -> Iterator[dict]:
        """Iterate over list of account subscriptions with optional filtering.

        Note:
            The name of this method is based on the API's method name.
            This method provides iteration over subscriptions, it does
            not return a list.

        Args:
            created (str): filter by created time or interval.
            end_time (str): filter by end time or interval.
            hosting (bool): only return subscriptions that contain a
                hosting block (e.g. SentinelHub hosting).
            name__contains (str): only return subscriptions with a
                name that contains the given string.
            name (str): filter by name.
            source_type (str): filter by source type.
            start_time (str): filter by start time or interval.
            status (Set[str]): include subscriptions with a status in this set.
            sort_by (str): fields to sort subscriptions by. Multiple
                fields can be specified, separated by commas. The sort
                direction can be specified by appending ' ASC' or '
                DESC' to the field name. The default sort direction is
                ascending. When multiple fields are specified, the sort
                order is applied in the order the fields are listed.

                Supported fields: name, created, updated, start_time, end_time

                Examples:
                 * "name"
                 * "name DESC"
                 * "name,end_time DESC,start_time"
            updated (str): filter by updated time or interval.
            destination_ref (str): filter by subscriptions created with the
                provided destination reference.
            user_id (str or int): filter by user ID. Only available to organization admins.
                Accepts "all" or a specific user ID.
            geom_ref (str): A feature reference to filter by.
            limit (int): limit the number of subscriptions in the
                results. When set to 0, no maximum is applied.
            page_size (int): number of subscriptions to return per page.

        Datetime args (created, end_time, start_time, updated) can either be a
        date-time or an interval, open or closed. Date and time expressions adhere
        to RFC 3339. Open intervals are expressed using double-dots.

        Examples:
            * A date-time: "2018-02-12T23:20:50Z"
            * A closed interval: "2018-02-12T00:00:00Z/2018-03-18T12:31:12Z"
            * Open intervals: "2018-02-12T00:00:00Z/.." or "../2018-03-18T12:31:12Z"

        Yields:
            dict: a description of a subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """

        return self._client._aiter_to_iter(
            self._client.list_subscriptions(status,
                                            limit,
                                            created,
                                            end_time,
                                            hosting,
                                            name__contains,
                                            name,
                                            source_type,
                                            start_time,
                                            sort_by,
                                            updated,
                                            destination_ref,
                                            user_id,
                                            geom_ref,
                                            page_size))

    def create_subscription(self, request: Dict) -> Dict:
        """Create a Subscription.

        Args:
            request (dict): description of a subscription.

        Returns:
            dict: description of created subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        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.

        Args:
            subscription_id (str): id of subscription to cancel.

        Returns:
            None

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.cancel_subscription(subscription_id))

    def suspend_subscription(self,
                             subscription_id: str,
                             details: Optional[str] = None) -> dict:
        """Suspend a Subscription.

        Args:
            subscription_id (str): id of subscription to suspend.
            details (str): optional details explaining the reason
                for suspension.

        Returns:
            dict: the suspended subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.suspend_subscription(subscription_id, details))

    def reactivate_subscription(self, subscription_id: str) -> None:
        """Reactivate a Subscription.

        Args:
            subscription_id (str): id of subscription to reactivate.

        Returns:
            None

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.reactivate_subscription(subscription_id))

    def bulk_suspend_subscriptions(self,
                                   subscription_ids: Optional[
                                       List[str]] = None,
                                   details: Optional[str] = None,
                                   all_subscriptions: bool = False) -> None:
        """Suspend multiple Subscriptions.

        This method supports three modes of operation:

        1. Suspend specific subscriptions: provide subscription_ids
        2. Suspend all user's subscriptions: call with no parameters
        3. Suspend all organization subscriptions: set all_subscriptions=True
           (organization admin only)

        Args:
            subscription_ids (List[str]): list of subscription ids
                to suspend. If not provided and all_subscriptions is False,
                suspends all of the user's subscriptions.
            details (str): optional details explaining the reason
                for suspension.
            all_subscriptions (bool): if True, suspend all
                subscriptions for the organization (requires organization
                admin permissions). Mutually exclusive with subscription_ids.

        Returns:
            None

        Raises:
            ClientError: if both subscription_ids and all_subscriptions are
                provided.
            APIError: on an API server error.
        """
        return self._client._call_sync(
            self._client.bulk_suspend_subscriptions(subscription_ids,
                                                    details,
                                                    all_subscriptions))

    def bulk_reactivate_subscriptions(self,
                                      subscription_ids: Optional[
                                          List[str]] = None,
                                      all_subscriptions: bool = False) -> None:
        """Reactivate multiple Subscriptions.

        This method supports three modes of operation:

        1. Reactivate specific subscriptions: provide subscription_ids
        2. Reactivate all user's subscriptions: call with no parameters
        3. Reactivate all organization subscriptions: set all_subscriptions=True
           (organization admin only)

        Args:
            subscription_ids (List[str]): list of subscription ids
                to reactivate. If not provided and all_subscriptions is False,
                reactivates all of the user's subscriptions.
            all_subscriptions (bool): if True, reactivate all
                subscriptions for the organization (requires organization
                admin permissions). Mutually exclusive with subscription_ids.

        Returns:
            None

        Raises:
            ClientError: if both subscription_ids and all_subscriptions are
                provided.
            APIError: on an API server error.
        """
        return self._client._call_sync(
            self._client.bulk_reactivate_subscriptions(subscription_ids,
                                                       all_subscriptions))

    def update_subscription(self, subscription_id: str, request: dict) -> dict:
        """Update (edit) a Subscription via PUT.

        Args:
            subscription_id (str): id of the subscription to update.
            request (dict): subscription content for update, full
                payload is required.

        Returns:
            dict: description of the updated subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.update_subscription(subscription_id, request))

    def patch_subscription(self, subscription_id: str,
                           request: Dict[str, Any]) -> Dict[str, Any]:
        """Update (edit) a Subscription via PATCH.

        Args:
            subscription_id (str): id of the subscription to update.
            request (dict): subscription content for update, only
                attributes to update are required.

        Returns:
            dict: description of the updated subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.patch_subscription(subscription_id, request))

    def get_subscription(self, subscription_id: str) -> Dict[str, Any]:
        """Get a description of a Subscription.

        Args:
            subscription_id (str): id of a subscription.

        Returns:
            dict: description of the subscription.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.get_subscription(subscription_id))

    def get_results(
        self,
        subscription_id: str,
        status: Optional[Sequence[Literal["created",
                                          "queued",
                                          "processing",
                                          "failed",
                                          "success"]]] = None,
        limit: int = 100,
        created: Optional[str] = None,
        updated: Optional[str] = None,
        completed: Optional[str] = None,
        item_datetime: Optional[str] = None
    ) -> Iterator[Union[Dict[str, Any], str]]:
        """Iterate over results of a Subscription.

        Notes:
            The name of this method is based on the API's method name. This
            method provides iteration over results, it does not get a
            single result description or return a list of descriptions.

        Parameters:
            subscription_id (str): id of a subscription.
            status (Set[str]): pass result with status in this set,
                filter out results with status not in this set.
            limit (int): limit the number of subscriptions in the
                results. When set to 0, no maximum is applied.
            created (str): filter by created time or interval.
            updated (str): filter by updated time or interval.
            completed (str): filter by completed time or interval.
            item_datetime (str): filter by item datetime or interval.

        Datetime args (created, updated, completed, item_datetime) can either be a
        date-time or an interval, open or closed. Date and time expressions adhere
        to RFC 3339. Open intervals are expressed using double-dots.

        Examples:
            * A date-time: "2018-02-12T23:20:50Z"
            * A closed interval: "2018-02-12T00:00:00Z/2018-03-18T12:31:12Z"
            * Open intervals: "2018-02-12T00:00:00Z/.." or "../2018-03-18T12:31:12Z"

        Yields:
            dict: description of a subscription results.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._aiter_to_iter(
            self._client.get_results(subscription_id,
                                     status,
                                     limit,
                                     created,
                                     updated,
                                     completed,
                                     item_datetime))

    def get_results_csv(self,
                        subscription_id: str,
                        status: Optional[Sequence[Literal["created",
                                                          "queued",
                                                          "processing",
                                                          "failed",
                                                          "success"]]] = None,
                        created: Optional[str] = None,
                        updated: Optional[str] = None,
                        completed: Optional[str] = None,
                        item_datetime: Optional[str] = None) -> Iterator[str]:
        """Iterate over rows of results CSV for a Subscription.

        Parameters:
            subscription_id (str): id of a subscription.
            status (Set[str]): pass result with status in this set,
                filter out results with status not in this set.
            created (str): filter by created time or interval.
            updated (str): filter by updated time or interval.
            completed (str): filter by completed time or interval.
            item_datetime (str): filter by item datetime or interval.

        Datetime args (created, updated, completed, item_datetime) can either be a
        date-time or an interval, open or closed. Date and time expressions adhere
        to RFC 3339. Open intervals are expressed using double-dots.

        Examples:
            * A date-time: "2018-02-12T23:20:50Z"
            * A closed interval: "2018-02-12T00:00:00Z/2018-03-18T12:31:12Z"
            * Open intervals: "2018-02-12T00:00:00Z/.." or "../2018-03-18T12:31:12Z"

        Yields:
            str: a row from a CSV file.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        # Note: retries are not implemented yet. This project has
        # retry logic for HTTP requests, but does not handle errors
        # during streaming. We may want to consider a retry decorator
        # for this entire method a la stamina:
        # https://github.com/hynek/stamina.
        return self._client._aiter_to_iter(
            self._client.get_results_csv(subscription_id,
                                         status,
                                         created,
                                         updated,
                                         completed,
                                         item_datetime))

    def get_summary(self) -> Dict[str, Any]:
        """Summarize the status of all subscriptions via GET.

        Returns:
            dict: subscription totals by status.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(self._client.get_summary())

    def get_subscription_summary(self, subscription_id: str) -> Dict[str, Any]:
        """Summarize the status of results for a single subscription via GET.

        Args
            subscription_id (str): ID of the subscription to summarize.

        Returns:
            dict: result totals for the provided subscription by status.

        Raises:
            APIError: on an API server error.
            ClientError: on a client error.
        """
        return self._client._call_sync(
            self._client.get_subscription_summary(subscription_id))