Module libnova.common.api.Driver

Expand source code
#!/usr/bin/env python
# coding: utf-8

import json
import os
import time
import sys
import hashlib
import requests
import urllib.request
import cysimdjson

from requests.structures import CaseInsensitiveDict
from requests            import Request, Session

__instance = None
__base_url = ''
__api_key  = ''


def get_instance():
    """Retrieve the Api Driver instance

    Returns:
        Driver: An Api `Driver`
    """

    global __instance
    global __base_url
    global __api_key

    if __instance is None:
        if __base_url == '' and __api_key == '':
            print('Initialize the driver first!')
        else:
            __instance = Driver(__base_url, __api_key)
    return __instance


def initialize(base_url, api_key, cleanup=False):
    """Initialize the Api Driver wrapper with the given endpoint and api key

    Args:
        base_url (str): The Api endpoint base url (without /api/)
        api_key (str): The Api key to use in each request
        cleanup (bool): Destroy the current instance if has been initialized previously
    """

    global __instance
    global __base_url
    global __api_key

    __base_url = base_url.rstrip('/')
    __api_key  = api_key

    if cleanup:
        __instance = None


class Driver:
    """Api Driver class

    This class is intended to be a wrapper in between the requests performed by the library against
    an Api endpoint

    This class handles the authentication, the response parsing and serializing if needed
    """

    base_url      = ''
    api_key       = ''
    proxy_enabled = False

    def __init__(self, base_url, key):
        self.base_url = base_url
        self.api_key  = key

    def set_api_key(self, key):
        """Set the current Api key

        Args:
            key (str): The Api key
        """

        self.api_key = key

    def set_base_url(self, base_url):
        """Set the current Api endpoint

        Args:
            base_url (str): The Api endpoint
        """

        self.base_url = str(base_url).rstrip('/')

    def get_url(self, segment):
        """Build an Api url based on a given segment

        Args:
            segment (str): The segment to append to the base Api endpoint url
        """

        return self.base_url + '/api/' + segment

    def get_headers(self, content_type=None):
        """Build the headers for an Api request

        Args:
            content_type (str): If set, it will be included in the headers as `Content-Type`
        """

        headers = {
            'Content-type':  content_type,
            'Authorization': 'Bearer ' + self.api_key
        }

        if content_type is not None:
            headers['Content-type'] = content_type
        return headers

    def set_proxy_enabled(self, enabled = True):
        """Set the Api client proxy status to the value of `enabled`

        Args:
            enabled (bool): True will enable the proxy (with automatic detection), False will disable it
        """

        self.proxy_enabled = enabled

    def get_proxy(self):
        """Return a list of proxies for an Api request

        Returns:
            dict: A dictionary with the different system proxies
        """

        if self.proxy_enabled:
            return urllib.request.getproxies()
        return None

    def get(self, url_segment, params=None, data=None):
        """Perform a GET request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.get(
            url,
            params=params,
            json=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)

    def print_curl_request(self, response):
        command = "curl -X {method} -H {headers} -d '{data}' '{uri}'"
        method  = response.request.method
        uri     = response.request.url
        data    = response.request.body
        headers = ['"{0}: {1}"'.format(k, v) for k, v in response.request.headers.items()]
        headers = " -H ".join(headers)

        print(command.format(method=method, headers=headers, data=data, uri=uri))

    def get_single(self, result):
        """Retrieve the first item of an Api response

        Args:
            result (dict): An Api response

        Returns:
            object: The first item in `result`; None if the `result` is empty or there is an error with the `result` format
        """

        if result is not None:
            try:
                return result[0]
            except Exception as e:
                # Do nothing, result is an array but is empty, so we should return None anyway
                pass
        return None

    def put(self, url_segment, params=None, data=None):
        """Perform a PUT request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.put(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)


    def delete(self, url_segment, params=None, data=None):
        """Perform a DELETE request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.delete(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)



    def post(self, url_segment, params=None, data=None):
        """Perform a POST request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.post(
            url,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)

    def post_json(self, url_segment, params=None, data=None):
        """Perform a POST request to the Api, but using the `data` as a JSON string

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The application/json input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        retries       = 0
        response_json = None
        while response_json is None and retries < 5:
            try:
                response = requests.post(
                    url,
                    json=data,
                    headers=self.get_headers('application/json'),
                    proxies=self.get_proxy(),
                    verify=not self.proxy_enabled
                )

                response_json = response.json()
            except:
                time.sleep(1)
                retries = retries + 1
        return self.__get_response_data(response_json)

    def put_json(self, url_segment, params=None, data=None):
        """Perform a PUT request to the Api, but using the `data` as a JSON string

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The application/json input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.put(
            url,
            json=data,
            headers=self.get_headers('application/json'),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        try:
            response_json = response.json()
            return self.__get_response_data(response_json)
        except Exception as ex:
            return {
                "success": False,
                "messages": [
                    ex
                ]
            }
            pass

    def __get_response_data(self, response_json):
        """Parse an Api response object and retrieve the set of results either as a list() or as a single object

        This behaviour depends on the Api endpoint itself

        Args:
            response_json (dict): An Api response

        Returns:
            object: A list() of items, a single item, or the raw Api response if there is an error
        """

        if response_json is not None:
            if 'success' in response_json:
                # Database query result
                if response_json['success'] == True:
                    # Elasticsearch query result
                    if 'hits' in response_json and 'hits' in response_json['hits']:
                        if len(response_json['hits']['hits']) > 0:
                            if len(response_json['hits']['hits']) == 1:
                                return response_json['hits']['hits'][0]['_source']
                            return list(map(lambda x : x['_source'], response_json['hits']['hits']))
                    if 'result' in response_json:
                        return response_json['result']

            # Return query result as is, as it holds the error message or the success status

        return response_json

    def serialize(self, response_json, data_type):
        """Serialize the results coming from `self.__get_response_data()` to list() or single `data_type` type

        Args:
            response_json (dict): A list of items coming from `self.__get_response_data()`
            data_type (class): A target class in which serialize each item

        Returns:
            object: A list() of items, a single item, or None if there is nothing to serialize to
        """

        if response_json is None:
            return None

        # Handle possible empty results returning from API when serializing to a structured class
        if "success" in response_json:
            if response_json["success"] == False:
                return None

        if isinstance(response_json, list):
            return list(map(lambda x: data_type(**x), response_json))
        else:
            return data_type(**response_json)

        # if isinstance(response_json, list):
        #     return list(map(lambda x: json.loads(x, object_hook=data_type), response_json))
        # else:
        #     return json.loads(response_json, object_hook= data_type)


if __name__ == "__main__":
    print('This file cannot be executed directly!')

Functions

def get_instance()

Retrieve the Api Driver instance

Returns

Driver
An Api Driver
Expand source code
def get_instance():
    """Retrieve the Api Driver instance

    Returns:
        Driver: An Api `Driver`
    """

    global __instance
    global __base_url
    global __api_key

    if __instance is None:
        if __base_url == '' and __api_key == '':
            print('Initialize the driver first!')
        else:
            __instance = Driver(__base_url, __api_key)
    return __instance
def initialize(base_url, api_key, cleanup=False)

Initialize the Api Driver wrapper with the given endpoint and api key

Args

base_url : str
The Api endpoint base url (without /api/)
api_key : str
The Api key to use in each request
cleanup : bool
Destroy the current instance if has been initialized previously
Expand source code
def initialize(base_url, api_key, cleanup=False):
    """Initialize the Api Driver wrapper with the given endpoint and api key

    Args:
        base_url (str): The Api endpoint base url (without /api/)
        api_key (str): The Api key to use in each request
        cleanup (bool): Destroy the current instance if has been initialized previously
    """

    global __instance
    global __base_url
    global __api_key

    __base_url = base_url.rstrip('/')
    __api_key  = api_key

    if cleanup:
        __instance = None

Classes

class Driver (base_url, key)

Api Driver class

This class is intended to be a wrapper in between the requests performed by the library against an Api endpoint

This class handles the authentication, the response parsing and serializing if needed

Expand source code
class Driver:
    """Api Driver class

    This class is intended to be a wrapper in between the requests performed by the library against
    an Api endpoint

    This class handles the authentication, the response parsing and serializing if needed
    """

    base_url      = ''
    api_key       = ''
    proxy_enabled = False

    def __init__(self, base_url, key):
        self.base_url = base_url
        self.api_key  = key

    def set_api_key(self, key):
        """Set the current Api key

        Args:
            key (str): The Api key
        """

        self.api_key = key

    def set_base_url(self, base_url):
        """Set the current Api endpoint

        Args:
            base_url (str): The Api endpoint
        """

        self.base_url = str(base_url).rstrip('/')

    def get_url(self, segment):
        """Build an Api url based on a given segment

        Args:
            segment (str): The segment to append to the base Api endpoint url
        """

        return self.base_url + '/api/' + segment

    def get_headers(self, content_type=None):
        """Build the headers for an Api request

        Args:
            content_type (str): If set, it will be included in the headers as `Content-Type`
        """

        headers = {
            'Content-type':  content_type,
            'Authorization': 'Bearer ' + self.api_key
        }

        if content_type is not None:
            headers['Content-type'] = content_type
        return headers

    def set_proxy_enabled(self, enabled = True):
        """Set the Api client proxy status to the value of `enabled`

        Args:
            enabled (bool): True will enable the proxy (with automatic detection), False will disable it
        """

        self.proxy_enabled = enabled

    def get_proxy(self):
        """Return a list of proxies for an Api request

        Returns:
            dict: A dictionary with the different system proxies
        """

        if self.proxy_enabled:
            return urllib.request.getproxies()
        return None

    def get(self, url_segment, params=None, data=None):
        """Perform a GET request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.get(
            url,
            params=params,
            json=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)

    def print_curl_request(self, response):
        command = "curl -X {method} -H {headers} -d '{data}' '{uri}'"
        method  = response.request.method
        uri     = response.request.url
        data    = response.request.body
        headers = ['"{0}: {1}"'.format(k, v) for k, v in response.request.headers.items()]
        headers = " -H ".join(headers)

        print(command.format(method=method, headers=headers, data=data, uri=uri))

    def get_single(self, result):
        """Retrieve the first item of an Api response

        Args:
            result (dict): An Api response

        Returns:
            object: The first item in `result`; None if the `result` is empty or there is an error with the `result` format
        """

        if result is not None:
            try:
                return result[0]
            except Exception as e:
                # Do nothing, result is an array but is empty, so we should return None anyway
                pass
        return None

    def put(self, url_segment, params=None, data=None):
        """Perform a PUT request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.put(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)


    def delete(self, url_segment, params=None, data=None):
        """Perform a DELETE request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.delete(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)



    def post(self, url_segment, params=None, data=None):
        """Perform a POST request to the Api

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The multipart/form-data input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.post(
            url,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        response_json = response.json()
        return self.__get_response_data(response_json)

    def post_json(self, url_segment, params=None, data=None):
        """Perform a POST request to the Api, but using the `data` as a JSON string

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The application/json input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        retries       = 0
        response_json = None
        while response_json is None and retries < 5:
            try:
                response = requests.post(
                    url,
                    json=data,
                    headers=self.get_headers('application/json'),
                    proxies=self.get_proxy(),
                    verify=not self.proxy_enabled
                )

                response_json = response.json()
            except:
                time.sleep(1)
                retries = retries + 1
        return self.__get_response_data(response_json)

    def put_json(self, url_segment, params=None, data=None):
        """Perform a PUT request to the Api, but using the `data` as a JSON string

        Args:
            url_segment (str): The Api segment to perform the request against
            params (dict): The url-encoded params to use in the request
            data (dict): The application/json input to use in the request

        Returns:
            dict: An Api response (see Documentation)
        """

        url = self.get_url(url_segment)

        response = requests.put(
            url,
            json=data,
            headers=self.get_headers('application/json'),
            proxies=self.get_proxy(),
            verify=not self.proxy_enabled
        )

        try:
            response_json = response.json()
            return self.__get_response_data(response_json)
        except Exception as ex:
            return {
                "success": False,
                "messages": [
                    ex
                ]
            }
            pass

    def __get_response_data(self, response_json):
        """Parse an Api response object and retrieve the set of results either as a list() or as a single object

        This behaviour depends on the Api endpoint itself

        Args:
            response_json (dict): An Api response

        Returns:
            object: A list() of items, a single item, or the raw Api response if there is an error
        """

        if response_json is not None:
            if 'success' in response_json:
                # Database query result
                if response_json['success'] == True:
                    # Elasticsearch query result
                    if 'hits' in response_json and 'hits' in response_json['hits']:
                        if len(response_json['hits']['hits']) > 0:
                            if len(response_json['hits']['hits']) == 1:
                                return response_json['hits']['hits'][0]['_source']
                            return list(map(lambda x : x['_source'], response_json['hits']['hits']))
                    if 'result' in response_json:
                        return response_json['result']

            # Return query result as is, as it holds the error message or the success status

        return response_json

    def serialize(self, response_json, data_type):
        """Serialize the results coming from `self.__get_response_data()` to list() or single `data_type` type

        Args:
            response_json (dict): A list of items coming from `self.__get_response_data()`
            data_type (class): A target class in which serialize each item

        Returns:
            object: A list() of items, a single item, or None if there is nothing to serialize to
        """

        if response_json is None:
            return None

        # Handle possible empty results returning from API when serializing to a structured class
        if "success" in response_json:
            if response_json["success"] == False:
                return None

        if isinstance(response_json, list):
            return list(map(lambda x: data_type(**x), response_json))
        else:
            return data_type(**response_json)

Class variables

var api_key
var base_url
var proxy_enabled

Methods

def delete(self, url_segment, params=None, data=None)

Perform a DELETE request to the Api

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The multipart/form-data input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def delete(self, url_segment, params=None, data=None):
    """Perform a DELETE request to the Api

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The multipart/form-data input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    response = requests.delete(
        url,
        params=params,
        data=data,
        headers=self.get_headers(),
        proxies=self.get_proxy(),
        verify=not self.proxy_enabled
    )

    response_json = response.json()
    return self.__get_response_data(response_json)
def get(self, url_segment, params=None, data=None)

Perform a GET request to the Api

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The multipart/form-data input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def get(self, url_segment, params=None, data=None):
    """Perform a GET request to the Api

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The multipart/form-data input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    response = requests.get(
        url,
        params=params,
        json=data,
        headers=self.get_headers(),
        proxies=self.get_proxy(),
        verify=not self.proxy_enabled
    )

    response_json = response.json()
    return self.__get_response_data(response_json)
def get_headers(self, content_type=None)

Build the headers for an Api request

Args

content_type : str
If set, it will be included in the headers as Content-Type
Expand source code
def get_headers(self, content_type=None):
    """Build the headers for an Api request

    Args:
        content_type (str): If set, it will be included in the headers as `Content-Type`
    """

    headers = {
        'Content-type':  content_type,
        'Authorization': 'Bearer ' + self.api_key
    }

    if content_type is not None:
        headers['Content-type'] = content_type
    return headers
def get_proxy(self)

Return a list of proxies for an Api request

Returns

dict
A dictionary with the different system proxies
Expand source code
def get_proxy(self):
    """Return a list of proxies for an Api request

    Returns:
        dict: A dictionary with the different system proxies
    """

    if self.proxy_enabled:
        return urllib.request.getproxies()
    return None
def get_single(self, result)

Retrieve the first item of an Api response

Args

result : dict
An Api response

Returns

object
The first item in result; None if the result is empty or there is an error with the result format
Expand source code
def get_single(self, result):
    """Retrieve the first item of an Api response

    Args:
        result (dict): An Api response

    Returns:
        object: The first item in `result`; None if the `result` is empty or there is an error with the `result` format
    """

    if result is not None:
        try:
            return result[0]
        except Exception as e:
            # Do nothing, result is an array but is empty, so we should return None anyway
            pass
    return None
def get_url(self, segment)

Build an Api url based on a given segment

Args

segment : str
The segment to append to the base Api endpoint url
Expand source code
def get_url(self, segment):
    """Build an Api url based on a given segment

    Args:
        segment (str): The segment to append to the base Api endpoint url
    """

    return self.base_url + '/api/' + segment
def post(self, url_segment, params=None, data=None)

Perform a POST request to the Api

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The multipart/form-data input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def post(self, url_segment, params=None, data=None):
    """Perform a POST request to the Api

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The multipart/form-data input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    response = requests.post(
        url,
        data=data,
        headers=self.get_headers(),
        proxies=self.get_proxy(),
        verify=not self.proxy_enabled
    )

    response_json = response.json()
    return self.__get_response_data(response_json)
def post_json(self, url_segment, params=None, data=None)

Perform a POST request to the Api, but using the data as a JSON string

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The application/json input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def post_json(self, url_segment, params=None, data=None):
    """Perform a POST request to the Api, but using the `data` as a JSON string

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The application/json input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    retries       = 0
    response_json = None
    while response_json is None and retries < 5:
        try:
            response = requests.post(
                url,
                json=data,
                headers=self.get_headers('application/json'),
                proxies=self.get_proxy(),
                verify=not self.proxy_enabled
            )

            response_json = response.json()
        except:
            time.sleep(1)
            retries = retries + 1
    return self.__get_response_data(response_json)
def print_curl_request(self, response)
Expand source code
def print_curl_request(self, response):
    command = "curl -X {method} -H {headers} -d '{data}' '{uri}'"
    method  = response.request.method
    uri     = response.request.url
    data    = response.request.body
    headers = ['"{0}: {1}"'.format(k, v) for k, v in response.request.headers.items()]
    headers = " -H ".join(headers)

    print(command.format(method=method, headers=headers, data=data, uri=uri))
def put(self, url_segment, params=None, data=None)

Perform a PUT request to the Api

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The multipart/form-data input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def put(self, url_segment, params=None, data=None):
    """Perform a PUT request to the Api

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The multipart/form-data input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    response = requests.put(
        url,
        params=params,
        data=data,
        headers=self.get_headers(),
        proxies=self.get_proxy(),
        verify=not self.proxy_enabled
    )

    response_json = response.json()
    return self.__get_response_data(response_json)
def put_json(self, url_segment, params=None, data=None)

Perform a PUT request to the Api, but using the data as a JSON string

Args

url_segment : str
The Api segment to perform the request against
params : dict
The url-encoded params to use in the request
data : dict
The application/json input to use in the request

Returns

dict
An Api response (see Documentation)
Expand source code
def put_json(self, url_segment, params=None, data=None):
    """Perform a PUT request to the Api, but using the `data` as a JSON string

    Args:
        url_segment (str): The Api segment to perform the request against
        params (dict): The url-encoded params to use in the request
        data (dict): The application/json input to use in the request

    Returns:
        dict: An Api response (see Documentation)
    """

    url = self.get_url(url_segment)

    response = requests.put(
        url,
        json=data,
        headers=self.get_headers('application/json'),
        proxies=self.get_proxy(),
        verify=not self.proxy_enabled
    )

    try:
        response_json = response.json()
        return self.__get_response_data(response_json)
    except Exception as ex:
        return {
            "success": False,
            "messages": [
                ex
            ]
        }
        pass
def serialize(self, response_json, data_type)

Serialize the results coming from self.__get_response_data() to list() or single data_type type

Args

response_json : dict
A list of items coming from self.__get_response_data()
data_type : class
A target class in which serialize each item

Returns

object
A list() of items, a single item, or None if there is nothing to serialize to
Expand source code
def serialize(self, response_json, data_type):
    """Serialize the results coming from `self.__get_response_data()` to list() or single `data_type` type

    Args:
        response_json (dict): A list of items coming from `self.__get_response_data()`
        data_type (class): A target class in which serialize each item

    Returns:
        object: A list() of items, a single item, or None if there is nothing to serialize to
    """

    if response_json is None:
        return None

    # Handle possible empty results returning from API when serializing to a structured class
    if "success" in response_json:
        if response_json["success"] == False:
            return None

    if isinstance(response_json, list):
        return list(map(lambda x: data_type(**x), response_json))
    else:
        return data_type(**response_json)
def set_api_key(self, key)

Set the current Api key

Args

key : str
The Api key
Expand source code
def set_api_key(self, key):
    """Set the current Api key

    Args:
        key (str): The Api key
    """

    self.api_key = key
def set_base_url(self, base_url)

Set the current Api endpoint

Args

base_url : str
The Api endpoint
Expand source code
def set_base_url(self, base_url):
    """Set the current Api endpoint

    Args:
        base_url (str): The Api endpoint
    """

    self.base_url = str(base_url).rstrip('/')
def set_proxy_enabled(self, enabled=True)

Set the Api client proxy status to the value of enabled

Args

enabled : bool
True will enable the proxy (with automatic detection), False will disable it
Expand source code
def set_proxy_enabled(self, enabled = True):
    """Set the Api client proxy status to the value of `enabled`

    Args:
        enabled (bool): True will enable the proxy (with automatic detection), False will disable it
    """

    self.proxy_enabled = enabled