Source code for civis.base

from builtins import super
import os
from posixpath import join
import threading
from concurrent import futures
import warnings
from requests import Request

from civis.response import PaginatedResponse, convert_response_data_type
from civis._utils import open_session, retry_request, MAX_RETRIES

FINISHED = ['success', 'succeeded']
FAILED = ['failed']
NOT_FINISHED = ['queued', 'running']
CANCELLED = ['cancelled']

# Translate Civis state strings into `future` state strings
for name in FINISHED + FAILED:
    STATE_TRANS[name] = futures._base.FINISHED
for name in NOT_FINISHED:
    STATE_TRANS[name] = futures._base.RUNNING
for name in CANCELLED:
    STATE_TRANS[name] = futures._base.CANCELLED_AND_NOTIFIED


def tostr_urljoin(*x):
    return join(*map(str, x))

class CivisJobFailure(Exception):
    def __init__(self, err_msg, response=None):
        self.error_message = err_msg
        self.response = response

    def __str__(self):
        return self.error_message

class CivisAPIError(Exception):
    def __init__(self, response):
        if response.content:  # the API itself gave an error response
            json = response.json()
            self.error_message = json["errorDescription"]
        else:  # this was something like a 502
            self.error_message = response.reason

        self.status_code = response.status_code
        self._response = response

    def __str__(self):
        if self.status_code:
            return "({}) {}".format(self.status_code, self.error_message)
            return self.error_message

class EmptyResultError(Exception):

class CivisAPIKeyError(Exception):

class CivisImportError(Exception):

def get_base_url():
    base_url = os.environ.get('CIVIS_API_ENDPOINT', DEFAULT_API_ENDPOINT)
    if not base_url.endswith('/'):
        base_url += '/'
    return base_url

class Endpoint(object):

    _lock = threading.Lock()

    def __init__(self, session_kwargs, client, return_type='civis'):
        self._session_kwargs = session_kwargs
        self._return_type = return_type
        self._base_url = get_base_url()
        self._client = client

    def _build_path(self, path):
        if not path:
            return self._base_url
        return tostr_urljoin(self._base_url, path.strip("/"))

    def _make_request(self, method, path=None, params=None, data=None,
        url = self._build_path(path)

        with self._lock:
            with open_session(**self._session_kwargs) as sess:
                request = Request(method, url, json=data,
                                  params=params, **kwargs)
                pre_request = sess.prepare_request(request)
                response = retry_request(method, pre_request,
                                         sess, MAX_RETRIES)

        if response.status_code == 401:
            auth_error = response.headers["www-authenticate"]
            raise CivisAPIKeyError(auth_error) from CivisAPIError(response)

        if not response.ok:
            raise CivisAPIError(response)

        return response

    def _call_api(self, method, path=None, params=None, data=None, **kwargs):
        iterator = kwargs.pop('iterator', False)

        if iterator:
            resp = PaginatedResponse(path, params, self)
            resp = self._make_request(method, path, params, data, **kwargs)
            resp = convert_response_data_type(resp,
        self._client.last_response = resp
        return resp

class CivisAsyncResultBase(futures.Future):
    """A base class for tracking asynchronous results.

    Sub-classes needs to call either the `set_result` method to set a result
    when it finishes or call `set_exception` if there is an error.

    The `_result` attribute can also be set to change the current state of
    the result. The `_result_` object needs to be set to an object with a
    ``state`` attribute. Alternatively the `_check_result` method can be
    overwritten to change how the state of the object is returned.

    poller : func
        A function which returns an object that has a ``state`` attribute.
    poller_args : tuple
        The arguments with which to call the poller function.
    polling_interval : int or float
        The number of seconds between API requests to check whether a result
        is ready.
    api_key : DEPRECATED str, optional
        Your Civis API key. If not given, the :envvar:`CIVIS_API_KEY`
        environment variable will be used.
    client : :class:`civis.APIClient`, optional
        If not provided, an :class:`civis.APIClient` object will be
        created from the :envvar:`CIVIS_API_KEY`.
    poll_on_creation : bool, optional
        If ``True`` (the default), it will poll upon calling ``result()`` the
        first time. If ``False``, it will wait the number of seconds specified
        in `polling_interval` from object creation before polling.
    def __init__(self, poller, poller_args,
                 polling_interval=None, api_key=None, client=None,
        self.poller = poller
        self.poller_args = poller_args
        self.polling_interval = polling_interval
        if api_key is not None:
            warnings.warn('The "api_key" parameter is deprecated and will be '
                          'removed in v2. Please use the `client` parameter '
                          'instead.', FutureWarning)
        self.client = client
        self.poll_on_creation = poll_on_creation

    def __repr__(self):
        # Almost the same as the superclass's __repr__, except we use
        # the `_civis_state` rather than the `_state`.
        with self._condition:
            if self._civis_state in FINISHED + FAILED:
                if self.exception():
                    return '<%s at %#x state=%s raised %s>' % (
                    return '<%s at %#x state=%s returned %s>' % (
            out = '<%s at %#x state=%s>' % (self.__class__.__name__,
            return out

    def cancel(self):
        """Not currently implemented."""
        raise NotImplementedError("Running jobs cannot currently be cancelled")

    def succeeded(self):
        """Return ``True`` if the job completed in Civis with no error."""
        with self._condition:
            return self._civis_state in FINISHED

    def failed(self):
        """Return ``True`` if the Civis job failed."""
        with self._condition:
            return self._civis_state in FAILED

    def _check_result(self):
        with self._condition:
            if self._result is not None:
                return self._result

    def _civis_state(self):
        """State as returned from Civis."""
        with self._condition:
            if self._check_result():
                return self._check_result().state
            return 'running'

    def _state(self):
        """State of the CivisAsyncResultBase in `future` language."""
        with self._condition:
            return STATE_TRANS[self._civis_state]

    def _state(self, value):
        # Ignore attempts to set the _state from the `Future` superclass

    def set_result(self, result):
        """Sets the return value of work associated with the future.

        This is adapted from
        This version does not try to change the _state or check that the
        initial _state is running since the Civis implementation has _state
        depend on the Platform job state.
        with self._condition:
            self._result = result
            for waiter in self._waiters:

    def set_exception(self, exception):
        """Sets the result of the future as being the given exception.

        This is adapted from
        This version does not try to change the _state or check that the
        initial _state is running since the Civis implementation has _state
        depend on the Platform job state.
        with self._condition:
            self._exception = exception
            for waiter in self._waiters: