Source code for bitex.interface.base

"""Base class for interfaces."""
# Import Built-Ins
import logging
import abc

# Import Third-Party

# Import Homebrew

# Init Logging Facilities
log = logging.getLogger(__name__)


[docs]class Interface(metaclass=abc.ABCMeta): """Base class for Interface objects.""" def __init__(self, *, name, rest_api): """ Initialize the Interface class instance. :param name: str, name of the Interface :param rest_api: REST API object. """ self.REST = rest_api # pylint: disable=invalid-name self.name = name try: self._supported_pairs = self._get_supported_pairs() except NotImplementedError: self._supported_pairs = None @property def supported_pairs(self): """ Return a list of supported currncy pairs. :return: list """ return self._supported_pairs @abc.abstractmethod def _get_supported_pairs(self): """ Generate a list of supported pairs. Queries the API for a list of supported pairs and returns this as a list. Raises a NotImplementedError by default and needs to be overridden in child classes. :raises: NotImplementedError """ raise NotImplementedError
[docs] def is_supported(self, pair): """ Check if the given pair is present in self._supported_pairs. Input can either be a string or a PairFormatter Obj (or child thereof). If it is the latter two, we'll call the format() method with the Interface's name attribute to acquire proper formatting. :param pair: Str, or bitex.pairs.PairFormatter Object :return: Bool """ try: pair = pair.format_for(self.name) except AttributeError: pair = pair if pair in self.supported_pairs: return True return False
[docs] def request(self, verb, endpoint, authenticate=False, **req_kwargs): """Query the API and return its result. :param verb: HTTP verb (GET, PUT, DELETE, etc) :param endpoint: Str :param authenticate: Bool, whether to call private_query or public_query method. :param req_kwargs: Kwargs to pass to _query / :class:`requests.Request()` :raise: UnsupportedPairError :return: :class:`requests.Response() Obj` """ if authenticate: return self.REST.private_query(verb, endpoint, **req_kwargs) return self.REST.public_query(verb, endpoint, **req_kwargs)