Class SensorDatastoreService

st2common/st2common/services/datastore.py:324–362

DatastoreService class used by sensors. This class is meant to be used in context of long running processes (e.g. sensors) and it uses "create_token" method to generate a new auth token. A new token is also automatically generated once the old one expires. Note: This class does nee

class SensorDatastoreService(BaseDatastoreService):
    """
    DatastoreService class used by sensors. This class is meant to be used in context of long
    running processes (e.g. sensors) and it uses "create_token" method to generate a new auth
    token. A new token is also automatically generated once the old one expires.

    Note: This class does need database access (create_token method - to be able to create a new
    token).
    """

    def __init__(self, logger, pack_name, class_name, api_username):
        super(SensorDatastoreService, self).__init__(
            logger=logger, pack_name=pack_name, class_name=class_name
        )
        self._api_username = api_username
        self._token_expire = get_datetime_utc_now()

    def get_api_client(self):
        """
        Retrieve API client instance.
        """
        token_expire = self._token_expire <= get_datetime_utc_now()

        if not self._client or token_expire:
            # Note: Late import to avoid high import cost (time wise)
            from st2common.services.access import create_token

            self._logger.debug("Creating new Client object.")

            ttl = cfg.CONF.auth.service_token_ttl
            api_url = get_full_public_api_url()

            temporary_token = create_token(
                username=self._api_username, ttl=ttl, service=True
            )
            self._client = Client(api_url=api_url, token=temporary_token.token)
            self._token_expire = get_datetime_utc_now() + timedelta(seconds=ttl)

        return self._client