Module max_ard.admin

Perform admin functions

Provides

Administrative functions

  • Check data usage at the user or account level
  • add/delete/get/list stored user credentials
Expand source code
"""Perform admin functions

Provides
--------
  Administrative functions

  - Check data usage at the user or account level
  - add/delete/get/list stored user credentials

"""
import base64
import urllib
from datetime import datetime
from functools import wraps
from posixpath import join
from typing import Union

from pydantic import BaseModel

from max_ard.base_collections import (
    UsageArea,
    UsageAvailable,
    UsageCost,
    UsageLimits,
    hydrate,
)
from max_ard.exceptions import MissingARDResource
from max_ard.monitor import Monitor
from max_ard.session import ard_url, get_self, get_user_session, paginated_response

__all__ = ["AccountManager"]


class AdminUsage(BaseModel):
    """A model to hold account or user usage"""

    area: UsageArea
    cost: UsageCost
    limits: UsageLimits
    available: UsageAvailable

    def __str__(self):
        def u(val):
            if val == -1:
                return "unlimited"
            else:
                return f"{val:.2f}"

        return f"""
 Data Usage
 ----------

  Account limit: ${u(self.limits.annual_subscription_fee_limit)}

  Imagery ordered: ${self.cost.total_imagery_cost}
  ├ Fresh (< 90 days): {self.area.fresh_imagery_sqkm} sq.km (${self.cost.fresh_imagery_cost})
  ├ Standard (90 days - 3 years): {self.area.standard_imagery_sqkm} sq.km (${self.cost.standard_imagery_cost})
  └ Training (> 3 years): ${self.cost.training_imagery_cost}

  Remaining balance: ${u(self.available.total_imagery_balance)}
  
  """


class RegisteredCredentials(BaseModel):
    "A model to hold stored user credentials"
    credentials_id: str
    account_id: str
    description: Union[str, None]
    created: str
    modified: str

    def __repr__(self):
        if self.short_description not in [None, ""]:
            return f'<RegisteredCredentials "{self.credentials_id}" ({self.short_description})>'
        else:
            return f'<RegisteredCredentials "{self.credentials_id}">'

    @property
    def short_description(self):
        if self.description is None:
            return ""
        if len(self.description) > 20:
            return self.description[0:20] + "..."
        else:
            return self.description


class AccountManager:
    """Manages account-related actions

    Arguments
    ---------
    account_id : str (optional)
        Account ID to manage. If not provided, will use the current user's account
    session : Requests session object (optional)
        The session to use to communicate with the API. If not provided, uses the current user's authenticated session
    """

    __all__ = ["get_account_usage", "get_user_usage"]

    def __init__(self, account_id=None, session=None) -> None:
        if not session:
            session = get_user_session()
        self.session = session
        if not account_id:
            account_id = get_self(self.session)["user"]["account_id"]
        self.account_id = account_id

    def admin_url(self, *args):
        return ard_url("admin", "account", self.account_id, *args)

    @property
    def properties(self):
        return self.session.get(self.admin_url()).json()["account"]

    ########
    # Usage
    ########

    def _validate_dates(self, *args):
        """Validate that dates are YYYY-MM-DD

        Arguments
        ---------
        *args : str
            One or more date strings to validate

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If one or more dates was not YYYY-MM-DD"""

        for date in args:
            if date not in [None, ""]:
                try:
                    datetime.strptime(date, "%Y-%m-%d")
                except ValueError:
                    raise ValueError("Dates must be YYYY-MM-DD")

    def get_account_usage(self, start_date=None, end_date=None):
        """Get account-level data usage

        Parameters
        ----------
        start_date : str (optional)
            Start date to calculate usage
        start_date : str (optional)
            End date to calculate usage

        Returns
        -------
        AdminUsage
            Object model of data usage
        """

        url = ard_url("usage", "account", self.account_id)

        self._validate_dates(start_date, end_date)

        params = {"start_date": start_date, "end_date": end_date}

        r = self.session.get(url, params=params)
        return AdminUsage(**r.json()["usage"])

    def get_user_usage(self, username=None, start_date=None, end_date=None):
        """Get user-level data usage

        Parameters
        ----------
        username : str (optional)
            User name to calculate usage for, if not provided the current user is used
        start_date : str (optional)
            Start date to calculate usage
        end_date : str (optional)
            End date to calculate usage

        Returns
        -------
        AdminUsage
            Object model of data usage
        """

        if not username:
            username = get_self(self.session)["user"]["user_id"]

        url = ard_url("usage", "user", username)

        self._validate_dates(start_date, end_date)

        params = {"start_date": start_date, "end_date": end_date}

        r = self.session.get(url, params=params)
        return AdminUsage(**r.json()["usage"])

    #####
    # Credentials Storage
    #####

    def add_credentials(self, name, credentials, description=""):
        """Saves credentials in the Crendentials API

        The stored credentials key (SAS Url for Azure or Base64-encoded Credentials JSON for GCS) is never returned.
        Only the name, description, and other metadata about the credentials object is provided.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the stored credentials, should have URL-friendly characters (no spaces)
        credentials : str
            Credentials secret to store (SAS Url for Azure or Base64 Credentials JSON for GCS).
            If this is a path to a GCS JSON file, it will be read and Base64 encoded.
        description : str (optional)
            Description of the credentials

        Returns
        -------
        dict
            API response JSON dictionary
        """

        if urllib.parse.quote(name) != name:
            raise ValueError(
                "Stored credential names should not use characters that require URL-encoding"
            )
        url = self.admin_url("credentials", name)

        try:
            with open(credentials, "rb") as binary_file:
                encoded = base64.b64encode(binary_file.read())
                credentials = encoded.decode("ascii")

        except (FileNotFoundError, OSError):
            pass

        payload = {"credentials": credentials, "description": description}
        response = self.session.put(url, json=payload)

        if response.status_code == 204:
            return None
        else:
            return response.json()

    def get_credentials(self, name, raw=False):
        """Retrieves information about stored credentials.

        The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
        only the name, description, and other metadata about the credentials object.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the stored credentials
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        RegisteredCredentials
            Object model of credentials
        """

        url = self.admin_url("credentials", name)
        response = self.session.get(url)
        if raw:
            return response.json()
        else:
            return RegisteredCredentials(**response.json()["registered_credentials"])

    def list_credentials(self, raw=False):
        """Retrieves all stored credentials in an account.

        The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
        only the name, description, and other metadata about the credentials object.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        List of RegisteredCredentials
            List of object models of credentials
        """

        # responses can be paginated
        id_key = lambda credential: credential["registered_credentials"]["credentials_id"]
        url = self.admin_url("credentials")

        credentials = paginated_response(self.session, url, id_key)

        if raw:
            return credentials
        else:
            object_key = lambda credential: credential["registered_credentials"]
            return hydrate(RegisteredCredentials, credentials, key=object_key)

    def delete_credentials(self, name):
        """Deletes stored credentials by name.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the credential object

        """
        url = self.admin_url("credentials", name)
        return self.session.delete(url)

    #####
    # Monitors
    #####

    def list_monitors(self, raw=False):
        """Retrieves all monitors for an account.

        Parameters
        ----------
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        List of Monitors
            List of object models of monitors
        """

        id_key = lambda monitor: monitor["monitor_id"]

        url = self.admin_url("monitor")
        monitors = paginated_response(self.session, url, id_key)

        if raw:
            return monitors
        else:
            return [Monitor.from_response(monitor) for monitor in monitors]

Classes

class AccountManager (account_id=None, session=None)

Manages account-related actions

Arguments

account_id : str (optional)
Account ID to manage. If not provided, will use the current user's account
session : Requests session object (optional)
The session to use to communicate with the API. If not provided, uses the current user's authenticated session
Expand source code
class AccountManager:
    """Manages account-related actions

    Arguments
    ---------
    account_id : str (optional)
        Account ID to manage. If not provided, will use the current user's account
    session : Requests session object (optional)
        The session to use to communicate with the API. If not provided, uses the current user's authenticated session
    """

    __all__ = ["get_account_usage", "get_user_usage"]

    def __init__(self, account_id=None, session=None) -> None:
        if not session:
            session = get_user_session()
        self.session = session
        if not account_id:
            account_id = get_self(self.session)["user"]["account_id"]
        self.account_id = account_id

    def admin_url(self, *args):
        return ard_url("admin", "account", self.account_id, *args)

    @property
    def properties(self):
        return self.session.get(self.admin_url()).json()["account"]

    ########
    # Usage
    ########

    def _validate_dates(self, *args):
        """Validate that dates are YYYY-MM-DD

        Arguments
        ---------
        *args : str
            One or more date strings to validate

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If one or more dates was not YYYY-MM-DD"""

        for date in args:
            if date not in [None, ""]:
                try:
                    datetime.strptime(date, "%Y-%m-%d")
                except ValueError:
                    raise ValueError("Dates must be YYYY-MM-DD")

    def get_account_usage(self, start_date=None, end_date=None):
        """Get account-level data usage

        Parameters
        ----------
        start_date : str (optional)
            Start date to calculate usage
        start_date : str (optional)
            End date to calculate usage

        Returns
        -------
        AdminUsage
            Object model of data usage
        """

        url = ard_url("usage", "account", self.account_id)

        self._validate_dates(start_date, end_date)

        params = {"start_date": start_date, "end_date": end_date}

        r = self.session.get(url, params=params)
        return AdminUsage(**r.json()["usage"])

    def get_user_usage(self, username=None, start_date=None, end_date=None):
        """Get user-level data usage

        Parameters
        ----------
        username : str (optional)
            User name to calculate usage for, if not provided the current user is used
        start_date : str (optional)
            Start date to calculate usage
        end_date : str (optional)
            End date to calculate usage

        Returns
        -------
        AdminUsage
            Object model of data usage
        """

        if not username:
            username = get_self(self.session)["user"]["user_id"]

        url = ard_url("usage", "user", username)

        self._validate_dates(start_date, end_date)

        params = {"start_date": start_date, "end_date": end_date}

        r = self.session.get(url, params=params)
        return AdminUsage(**r.json()["usage"])

    #####
    # Credentials Storage
    #####

    def add_credentials(self, name, credentials, description=""):
        """Saves credentials in the Crendentials API

        The stored credentials key (SAS Url for Azure or Base64-encoded Credentials JSON for GCS) is never returned.
        Only the name, description, and other metadata about the credentials object is provided.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the stored credentials, should have URL-friendly characters (no spaces)
        credentials : str
            Credentials secret to store (SAS Url for Azure or Base64 Credentials JSON for GCS).
            If this is a path to a GCS JSON file, it will be read and Base64 encoded.
        description : str (optional)
            Description of the credentials

        Returns
        -------
        dict
            API response JSON dictionary
        """

        if urllib.parse.quote(name) != name:
            raise ValueError(
                "Stored credential names should not use characters that require URL-encoding"
            )
        url = self.admin_url("credentials", name)

        try:
            with open(credentials, "rb") as binary_file:
                encoded = base64.b64encode(binary_file.read())
                credentials = encoded.decode("ascii")

        except (FileNotFoundError, OSError):
            pass

        payload = {"credentials": credentials, "description": description}
        response = self.session.put(url, json=payload)

        if response.status_code == 204:
            return None
        else:
            return response.json()

    def get_credentials(self, name, raw=False):
        """Retrieves information about stored credentials.

        The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
        only the name, description, and other metadata about the credentials object.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the stored credentials
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        RegisteredCredentials
            Object model of credentials
        """

        url = self.admin_url("credentials", name)
        response = self.session.get(url)
        if raw:
            return response.json()
        else:
            return RegisteredCredentials(**response.json()["registered_credentials"])

    def list_credentials(self, raw=False):
        """Retrieves all stored credentials in an account.

        The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
        only the name, description, and other metadata about the credentials object.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        List of RegisteredCredentials
            List of object models of credentials
        """

        # responses can be paginated
        id_key = lambda credential: credential["registered_credentials"]["credentials_id"]
        url = self.admin_url("credentials")

        credentials = paginated_response(self.session, url, id_key)

        if raw:
            return credentials
        else:
            object_key = lambda credential: credential["registered_credentials"]
            return hydrate(RegisteredCredentials, credentials, key=object_key)

    def delete_credentials(self, name):
        """Deletes stored credentials by name.

        This is only available to Admin users. However all users in the account may use a stored credential name
        to provide write access when ordering data.

        Parameters
        ----------
        name : str
            Name of the credential object

        """
        url = self.admin_url("credentials", name)
        return self.session.delete(url)

    #####
    # Monitors
    #####

    def list_monitors(self, raw=False):
        """Retrieves all monitors for an account.

        Parameters
        ----------
        raw : bool (optional)
            Return the response JSON

        Returns
        -------
        List of Monitors
            List of object models of monitors
        """

        id_key = lambda monitor: monitor["monitor_id"]

        url = self.admin_url("monitor")
        monitors = paginated_response(self.session, url, id_key)

        if raw:
            return monitors
        else:
            return [Monitor.from_response(monitor) for monitor in monitors]

Instance variables

var properties
Expand source code
@property
def properties(self):
    return self.session.get(self.admin_url()).json()["account"]

Methods

def add_credentials(self, name, credentials, description='')

Saves credentials in the Crendentials API

The stored credentials key (SAS Url for Azure or Base64-encoded Credentials JSON for GCS) is never returned. Only the name, description, and other metadata about the credentials object is provided.

This is only available to Admin users. However all users in the account may use a stored credential name to provide write access when ordering data.

Parameters

name : str
Name of the stored credentials, should have URL-friendly characters (no spaces)
credentials : str
Credentials secret to store (SAS Url for Azure or Base64 Credentials JSON for GCS). If this is a path to a GCS JSON file, it will be read and Base64 encoded.
description : str (optional)
Description of the credentials

Returns

dict
API response JSON dictionary
Expand source code
def add_credentials(self, name, credentials, description=""):
    """Saves credentials in the Crendentials API

    The stored credentials key (SAS Url for Azure or Base64-encoded Credentials JSON for GCS) is never returned.
    Only the name, description, and other metadata about the credentials object is provided.

    This is only available to Admin users. However all users in the account may use a stored credential name
    to provide write access when ordering data.

    Parameters
    ----------
    name : str
        Name of the stored credentials, should have URL-friendly characters (no spaces)
    credentials : str
        Credentials secret to store (SAS Url for Azure or Base64 Credentials JSON for GCS).
        If this is a path to a GCS JSON file, it will be read and Base64 encoded.
    description : str (optional)
        Description of the credentials

    Returns
    -------
    dict
        API response JSON dictionary
    """

    if urllib.parse.quote(name) != name:
        raise ValueError(
            "Stored credential names should not use characters that require URL-encoding"
        )
    url = self.admin_url("credentials", name)

    try:
        with open(credentials, "rb") as binary_file:
            encoded = base64.b64encode(binary_file.read())
            credentials = encoded.decode("ascii")

    except (FileNotFoundError, OSError):
        pass

    payload = {"credentials": credentials, "description": description}
    response = self.session.put(url, json=payload)

    if response.status_code == 204:
        return None
    else:
        return response.json()
def admin_url(self, *args)
Expand source code
def admin_url(self, *args):
    return ard_url("admin", "account", self.account_id, *args)
def delete_credentials(self, name)

Deletes stored credentials by name.

This is only available to Admin users. However all users in the account may use a stored credential name to provide write access when ordering data.

Parameters

name : str
Name of the credential object
Expand source code
def delete_credentials(self, name):
    """Deletes stored credentials by name.

    This is only available to Admin users. However all users in the account may use a stored credential name
    to provide write access when ordering data.

    Parameters
    ----------
    name : str
        Name of the credential object

    """
    url = self.admin_url("credentials", name)
    return self.session.delete(url)
def get_account_usage(self, start_date=None, end_date=None)

Get account-level data usage

Parameters

start_date : str (optional)
Start date to calculate usage
start_date : str (optional)
End date to calculate usage

Returns

AdminUsage
Object model of data usage
Expand source code
def get_account_usage(self, start_date=None, end_date=None):
    """Get account-level data usage

    Parameters
    ----------
    start_date : str (optional)
        Start date to calculate usage
    start_date : str (optional)
        End date to calculate usage

    Returns
    -------
    AdminUsage
        Object model of data usage
    """

    url = ard_url("usage", "account", self.account_id)

    self._validate_dates(start_date, end_date)

    params = {"start_date": start_date, "end_date": end_date}

    r = self.session.get(url, params=params)
    return AdminUsage(**r.json()["usage"])
def get_credentials(self, name, raw=False)

Retrieves information about stored credentials.

The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned, only the name, description, and other metadata about the credentials object.

This is only available to Admin users. However all users in the account may use a stored credential name to provide write access when ordering data.

Parameters

name : str
Name of the stored credentials
raw : bool (optional)
Return the response JSON

Returns

RegisteredCredentials
Object model of credentials
Expand source code
def get_credentials(self, name, raw=False):
    """Retrieves information about stored credentials.

    The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
    only the name, description, and other metadata about the credentials object.

    This is only available to Admin users. However all users in the account may use a stored credential name
    to provide write access when ordering data.

    Parameters
    ----------
    name : str
        Name of the stored credentials
    raw : bool (optional)
        Return the response JSON

    Returns
    -------
    RegisteredCredentials
        Object model of credentials
    """

    url = self.admin_url("credentials", name)
    response = self.session.get(url)
    if raw:
        return response.json()
    else:
        return RegisteredCredentials(**response.json()["registered_credentials"])
def get_user_usage(self, username=None, start_date=None, end_date=None)

Get user-level data usage

Parameters

username : str (optional)
User name to calculate usage for, if not provided the current user is used
start_date : str (optional)
Start date to calculate usage
end_date : str (optional)
End date to calculate usage

Returns

AdminUsage
Object model of data usage
Expand source code
def get_user_usage(self, username=None, start_date=None, end_date=None):
    """Get user-level data usage

    Parameters
    ----------
    username : str (optional)
        User name to calculate usage for, if not provided the current user is used
    start_date : str (optional)
        Start date to calculate usage
    end_date : str (optional)
        End date to calculate usage

    Returns
    -------
    AdminUsage
        Object model of data usage
    """

    if not username:
        username = get_self(self.session)["user"]["user_id"]

    url = ard_url("usage", "user", username)

    self._validate_dates(start_date, end_date)

    params = {"start_date": start_date, "end_date": end_date}

    r = self.session.get(url, params=params)
    return AdminUsage(**r.json()["usage"])
def list_credentials(self, raw=False)

Retrieves all stored credentials in an account.

The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned, only the name, description, and other metadata about the credentials object.

This is only available to Admin users. However all users in the account may use a stored credential name to provide write access when ordering data.

Parameters

raw : bool (optional)
Return the response JSON

Returns

List of RegisteredCredentials
List of object models of credentials
Expand source code
def list_credentials(self, raw=False):
    """Retrieves all stored credentials in an account.

    The stored credentials key (SAS Url for Azure or Credentials JSON for GCS) is never returned,
    only the name, description, and other metadata about the credentials object.

    This is only available to Admin users. However all users in the account may use a stored credential name
    to provide write access when ordering data.

    Parameters
    ----------
    raw : bool (optional)
        Return the response JSON

    Returns
    -------
    List of RegisteredCredentials
        List of object models of credentials
    """

    # responses can be paginated
    id_key = lambda credential: credential["registered_credentials"]["credentials_id"]
    url = self.admin_url("credentials")

    credentials = paginated_response(self.session, url, id_key)

    if raw:
        return credentials
    else:
        object_key = lambda credential: credential["registered_credentials"]
        return hydrate(RegisteredCredentials, credentials, key=object_key)
def list_monitors(self, raw=False)

Retrieves all monitors for an account.

Parameters

raw : bool (optional)
Return the response JSON

Returns

List of Monitors
List of object models of monitors
Expand source code
def list_monitors(self, raw=False):
    """Retrieves all monitors for an account.

    Parameters
    ----------
    raw : bool (optional)
        Return the response JSON

    Returns
    -------
    List of Monitors
        List of object models of monitors
    """

    id_key = lambda monitor: monitor["monitor_id"]

    url = self.admin_url("monitor")
    monitors = paginated_response(self.session, url, id_key)

    if raw:
        return monitors
    else:
        return [Monitor.from_response(monitor) for monitor in monitors]