Module `libnova.common.api.Driver`

Expand source code

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

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

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
    __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()
        )

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

    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.post(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy()
        )

        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()
        )

        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()
        )

        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)

        response = requests.post(
            url,
            json=data,
            headers=self.get_headers('application/json'),
            proxies=self.get_proxy()
        )

        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)
        """

        url = self.get_url(url_segment)

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

        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 '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
    __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()
        )

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

    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.post(
            url,
            params=params,
            data=data,
            headers=self.get_headers(),
            proxies=self.get_proxy()
        )

        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()
        )

        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()
        )

        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)

        response = requests.post(
            url,
            json=data,
            headers=self.get_headers('application/json'),
            proxies=self.get_proxy()
        )

        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)
        """

        url = self.get_url(url_segment)

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

        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 '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()
    )

    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()
    )

    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()
    )

    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)

    response = requests.post(
        url,
        json=data,
        headers=self.get_headers('application/json'),
        proxies=self.get_proxy()
    )

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

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.post(
        url,
        params=params,
        data=data,
        headers=self.get_headers(),
        proxies=self.get_proxy()
    )

    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()
    )

    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