# -*- coding: utf-8 -*- # import requests import re import types from functools import wraps from obsapi.null import Null from obsapi.logger import logger from requests.auth import HTTPBasicAuth try: import osc.conf as osc_conf except Exception: osc_conf = None DEFAULTAPIURL = 'https://api.opensuse.org' url_validate = re.compile(r'^(?:http|ftp)s?://' # http:// or https:// r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # domain... r'localhost|' # localhost... r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip r'(?::\d+)?' # optional port r'(?:/?|[/?]\S+)$', re.IGNORECASE).match class ObsHttpApi(object): """This class provides the lowest level OBS API HTTP interface This class provides basic interfaces to GET, PUT, POST calls to the Open Build Service API. The arguments to the get, put, post methods follow the schema of the OBS API URL paths. The class handles authentication using credentials setup with the `osc` command or falls back to .netrc. We don't do any real error checking here. Requests are sent to the OBS api server and response results returned. Callers are expected to interpret the results accordingly. The ObsHttpApi class is usually subclassed by higher level classes. Parameters ---------- apiurl: str Base url for the OBS api server (i.e. https://api.opensuse.org) Attributes ---------- apiurl (str): The base url for the OBS api server. (i.e. https://api.opensuse.org) verify_ssl (Boolean): Flag to verify ssl certificates """ default_xml = '' rootapi = '/' def __init__(self, apiurl=None): self._apiurl = DEFAULTAPIURL self.apiurl = apiurl self._auth = {} self._response = Null() self.retries = 0 self.verify_ssl = True self._callback_function = None def _with_callback(func): @wraps(func) def wrapper(inst, *args, **kwargs): r = func(inst, *args, **kwargs) inst._do_callback() return r return wrapper def _do_callback(self): if self._callback_function is not None: try: self._callback_function(self) except Exception as e: raise e def set_callback(self, func): """ Setup a callback function that is called after each HTTP request/response session to the OBS api server. This can be used by callers to trap and check conditions after calls (for example raising custom exceptions on HTTP errors). :param func: function to call """ if isinstance(func, (types.FunctionType, types.MethodType)): self._callback_function = func else: raise ValueError('Callback expects function or method type, ' 'got {}'.format(type(func))) @property def __auth(self): return self._auth.get(self.apiurl, self.__get_auth()) def __get_auth(self): conf = {} if osc_conf: try: osc_conf.get_config() conf = osc_conf.get_apiurl_api_host_options(self.apiurl) except Exception as e: logger.debug(e) pass user = conf.get('user', None) password = conf.get('pass', None) if user and password: self._auth[self.apiurl] = HTTPBasicAuth(user, password) else: self._auth[self.apiurl] = None return self._auth[self.apiurl] @_with_callback def __api_get(self, api, params=None): url = '{0}{1}{2}'.format(self.apiurl, self.rootapi, api) stream = params.pop('stream', False) def try_get(): """ """ r = requests.get(url, auth=self.__auth, params=params, stream=stream, verify=self.verify_ssl) self._response = r return r for attempt in range(1 + self.retries): r = try_get() if self.success: break return r @_with_callback def __api_put(self, api, data, params=None): url = '{0}{1}{2}'.format(self.apiurl, self.rootapi, api) r = requests.put(url, auth=self.__auth, data=data, params=params, verify=self.verify_ssl) self._response = r return r @_with_callback def __api_post(self, api, data, params=None): url = '{0}{1}{2}'.format(self.apiurl, self.rootapi, api) r = requests.post(url, auth=self.__auth, data=data, params=params, verify=self.verify_ssl) self._response = r return r @property def apiurl(self): """ str: The base url for the OBS api server. """ return self._apiurl @apiurl.setter def apiurl(self, url): if url_validate(url) is None: raise(Exception('Invalid URL: {}'.format(url))) self._apiurl = url @property def response(self): """python requests response obj: reference to the requests response from the last HTTP call.""" return self._response @property def success(self): """bool: True if last HTTP call was successful""" return self._response.status_code == requests.codes.ok def __path(self, *args): args = [elem for elem in args if elem] return '/'.join(['{}'] * len(args)).format(*args) def get(self, *args, **params): """ Performs an HTTP GET request to the OBS API. :param *args: arguments will be expanded to URL path elements. :param **params: kwargs are converted to HTTP query arguments :returns: content payload of the HTTP response as string or bytes. Example: >>> self.apiurl = 'https://api.opensuse.org' >>> get('build', 'myproj', 'openSUSE_Tumbleweed', 'x86_64', 'mypackage', 'filename', view='fileinfo') results in the following HTTP call GET "https://api.opensuse.org/build/myproj/openSUSE_Tumbleweed/x86_64/mypackage/filename?view=fileinfo" and returns the xml payload of the OBS response. """ binary_get = params.pop('binary_get', False) path = self.__path(*args) stream = params.get('stream', False) r = self.__api_get(path, params=params) if binary_get: if stream: return r return r.content return r.text def put(self, *args, **kwargs): """ Performs an HTTP PUT request to the OBS API. :param *args: arguments will be expanded to URL path elements. :param **params: kwargs are converted to HTTP query arguments :returns: content payload of the HTTP response as string or bytes. """ data = kwargs.pop('data', None) path = self.__path(*args) r = self.__api_put(path, data=data, params=kwargs) return r def post(self, *args, **kwargs): """ Performs an HTTP PUT request to the OBS API. :param *args: arguments will be expanded to URL path elements. :param **params: kwargs are converted to HTTP query arguments :returns: content payload of the HTTP response as string or bytes. """ data = kwargs.pop('data', None) path = self.__path(*args) r = self.__api_post(path, data=data, params=kwargs) return r