You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
405 lines
13 KiB
405 lines
13 KiB
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
flowroutenumbersandmessaging.api_helper
|
|
|
|
This file was automatically generated by APIMATIC v2.0 ( https://apimatic.io )
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
import datetime
|
|
import dateutil.parser
|
|
import calendar
|
|
import email.utils as eut
|
|
from time import mktime
|
|
|
|
import jsonpickle
|
|
import dateutil.parser
|
|
from requests.utils import quote
|
|
|
|
class APIHelper(object):
|
|
|
|
"""A Helper Class for various functions associated with API Calls.
|
|
|
|
This class contains static methods for operations that need to be
|
|
performed during API requests. All of the methods inside this class are
|
|
static methods, there is no need to ever initialise an instance of this
|
|
class.
|
|
|
|
"""
|
|
|
|
@staticmethod
|
|
def merge_dicts(dict1, dict2):
|
|
"""Merges two dictionaries into one as a shallow copy.
|
|
|
|
Args:
|
|
dict1 (dict): The first dictionary.
|
|
dict2 (dict): The second dictionary.
|
|
|
|
Returns:
|
|
dict: A dictionary containing key value pairs
|
|
from both the argument dictionaries. In the case
|
|
of a key conflict, values from dict2 are used
|
|
and those from dict1 are lost.
|
|
|
|
"""
|
|
temp = dict1.copy()
|
|
temp.update(dict2)
|
|
return temp
|
|
|
|
@staticmethod
|
|
def json_serialize(obj):
|
|
"""JSON Serialization of a given object.
|
|
|
|
Args:
|
|
obj (object): The object to serialise.
|
|
|
|
Returns:
|
|
str: The JSON serialized string of the object.
|
|
|
|
"""
|
|
if obj is None:
|
|
return None
|
|
|
|
# Resolve any Names if it's one of our objects that needs to have this called on
|
|
if isinstance(obj, list):
|
|
value = list()
|
|
for item in obj:
|
|
if hasattr(item, "_names"):
|
|
value.append(APIHelper.to_dictionary(item))
|
|
else:
|
|
value.append(item)
|
|
obj = value
|
|
else:
|
|
if hasattr(obj, "_names"):
|
|
obj = APIHelper.to_dictionary(obj)
|
|
|
|
return jsonpickle.encode(obj, False)
|
|
|
|
@staticmethod
|
|
def json_deserialize(json, unboxing_function=None):
|
|
"""JSON Deerialization of a given string.
|
|
|
|
Args:
|
|
json (str): The JSON serialized string to deserialize.
|
|
|
|
Returns:
|
|
dict: A dictionary representing the data contained in the
|
|
JSON serialized string.
|
|
|
|
"""
|
|
if json is None:
|
|
return None
|
|
|
|
try:
|
|
decoded = jsonpickle.decode(json)
|
|
except:
|
|
return json
|
|
|
|
if unboxing_function is None:
|
|
return decoded
|
|
elif isinstance(decoded, list):
|
|
return [unboxing_function(element) for element in decoded]
|
|
else:
|
|
return unboxing_function(decoded)
|
|
|
|
@staticmethod
|
|
def serialize_array(key, array, formatting="indexed"):
|
|
"""Converts an array parameter to a list of key value tuples.
|
|
|
|
Args:
|
|
key (str): The name of the parameter.
|
|
array (list): The value of the parameter.
|
|
formatting (str): The type of key formatting expected.
|
|
|
|
Returns:
|
|
list: A list with key value tuples for the array elements.
|
|
|
|
"""
|
|
tuples = []
|
|
|
|
if sys.version_info[0] < 3:
|
|
serializable_types = (str, int, long, float, bool, datetime.date, APIHelper.CustomDate)
|
|
else:
|
|
serializable_types = (str, int, float, bool, datetime.date, APIHelper.CustomDate)
|
|
|
|
if isinstance(array[0], serializable_types):
|
|
if formatting is "unindexed":
|
|
tuples += [("{0}[]".format(key), element) for element in array]
|
|
elif formatting is "indexed":
|
|
tuples += [("{0}[{1}]".format(key, index), element) for index, element in enumerate(array)]
|
|
elif formatting is "plain":
|
|
tuples += [(key, element) for element in array]
|
|
else:
|
|
raise ValueError("Invalid format provided.")
|
|
else:
|
|
tuples += [("{0}[{1}]".format(key, index), element) for index, element in enumerate(array)]
|
|
|
|
return tuples
|
|
|
|
@staticmethod
|
|
def append_url_with_template_parameters(url,
|
|
parameters):
|
|
"""Replaces template parameters in the given url.
|
|
|
|
Args:
|
|
url (str): The query url string to replace the template parameters.
|
|
parameters (dict): The parameters to replace in the url.
|
|
|
|
Returns:
|
|
str: URL with replaced parameters.
|
|
|
|
"""
|
|
# Parameter validation
|
|
if url is None:
|
|
raise ValueError("URL is None.")
|
|
if parameters is None:
|
|
return url
|
|
|
|
# Iterate and replace parameters
|
|
for key in parameters:
|
|
element = parameters[key]
|
|
replace_value = ""
|
|
|
|
# Load parameter value
|
|
if element is None:
|
|
replace_value = ""
|
|
elif isinstance(element, list):
|
|
replace_value = "/".join(quote(str(x), safe='') for x in element)
|
|
else:
|
|
replace_value = quote(str(element), safe='')
|
|
|
|
url = url.replace('{{{0}}}'.format(key), str(replace_value))
|
|
|
|
return url
|
|
|
|
@staticmethod
|
|
def append_url_with_query_parameters(url,
|
|
parameters,
|
|
array_serialization="indexed"):
|
|
"""Adds query parameters to a URL.
|
|
|
|
Args:
|
|
url (str): The URL string.
|
|
parameters (dict): The query parameters to add to the URL.
|
|
array_serialization (str): The format of array parameter serialization.
|
|
|
|
Returns:
|
|
str: URL with added query parameters.
|
|
|
|
"""
|
|
# Parameter validation
|
|
if url is None:
|
|
raise ValueError("URL is None.")
|
|
if parameters is None:
|
|
return url
|
|
|
|
for key, value in parameters.items():
|
|
seperator = '&' if '?' in url else '?'
|
|
if value is not None:
|
|
if isinstance(value, list):
|
|
value = [element for element in value if element]
|
|
if array_serialization is "csv":
|
|
url += "{0}{1}={2}".format(seperator, key,
|
|
",".join(quote(str(x), safe='') for x in value))
|
|
elif array_serialization is "psv":
|
|
url += "{0}{1}={2}".format(seperator, key,
|
|
"|".join(quote(str(x), safe='') for x in value))
|
|
elif array_serialization is "tsv":
|
|
url += "{0}{1}={2}".format(seperator, key,
|
|
"\t".join(quote(str(x), safe='') for x in value))
|
|
else:
|
|
url += "{0}{1}".format(seperator,
|
|
"&".join(("{0}={1}".format(k, quote(str(v), safe='')))
|
|
for k, v in APIHelper.serialize_array(key, value, array_serialization)))
|
|
else:
|
|
url += "{0}{1}={2}".format(seperator, key, quote(str(value), safe=''))
|
|
|
|
return url
|
|
|
|
@staticmethod
|
|
def clean_url(url):
|
|
"""Validates and processes the given query Url to clean empty slashes.
|
|
|
|
Args:
|
|
url (str): The given query Url to process.
|
|
|
|
Returns:
|
|
str: Clean Url as string.
|
|
|
|
"""
|
|
# Ensure that the urls are absolute
|
|
regex = "^https?://[^/]+"
|
|
match = re.match(regex, url)
|
|
if match is None:
|
|
raise ValueError('Invalid Url format.')
|
|
|
|
protocol = match.group(0)
|
|
index = url.find('?')
|
|
query_url = url[len(protocol): index if index != -1 else None]
|
|
query_url = re.sub("//+", "/", query_url)
|
|
parameters = url[index:] if index != -1 else ""
|
|
|
|
return protocol + query_url + parameters
|
|
|
|
@staticmethod
|
|
def form_encode_parameters(form_parameters,
|
|
array_serialization="indexed"):
|
|
"""Form encodes a dictionary of form parameters
|
|
|
|
Args:
|
|
form_parameters (dictionary): The given dictionary which has
|
|
atleast one model to form encode.
|
|
array_serialization (str): The format of array parameter serialization.
|
|
|
|
Returns:
|
|
dict: A dictionary of form encoded properties of the model.
|
|
|
|
"""
|
|
encoded = []
|
|
|
|
for key, value in form_parameters.items():
|
|
encoded += APIHelper.form_encode(value, key, array_serialization)
|
|
|
|
return encoded
|
|
|
|
@staticmethod
|
|
def form_encode(obj,
|
|
instance_name,
|
|
array_serialization="indexed"):
|
|
"""Encodes a model in a form-encoded manner such as person[Name]
|
|
|
|
Args:
|
|
obj (object): The given Object to form encode.
|
|
instance_name (string): The base name to appear before each entry
|
|
for this object.
|
|
array_serialization (string): The format of array parameter serialization.
|
|
|
|
Returns:
|
|
dict: A dictionary of form encoded properties of the model.
|
|
|
|
"""
|
|
retval = []
|
|
|
|
# If we received an object, resolve it's field names.
|
|
if hasattr(obj, "_names"):
|
|
obj = APIHelper.to_dictionary(obj)
|
|
|
|
if obj is None:
|
|
return []
|
|
elif isinstance(obj, list):
|
|
for element in APIHelper.serialize_array(instance_name, obj, array_serialization):
|
|
retval += APIHelper.form_encode(element[1], element[0], array_serialization)
|
|
elif isinstance(obj, dict):
|
|
for item in obj:
|
|
retval += APIHelper.form_encode(obj[item], instance_name + "[" + item + "]", array_serialization)
|
|
else:
|
|
retval.append((instance_name, obj))
|
|
|
|
return retval
|
|
|
|
@staticmethod
|
|
def to_dictionary(obj):
|
|
"""Creates a dictionary representation of a class instance. The
|
|
keys are taken from the API description and may differ from language
|
|
specific variable names of properties.
|
|
|
|
Args:
|
|
obj: The object to be converted into a dictionary.
|
|
|
|
Returns:
|
|
dictionary: A dictionary form of the model with properties in
|
|
their API formats.
|
|
|
|
"""
|
|
dictionary = dict()
|
|
|
|
# Loop through all properties in this model
|
|
for name in obj._names:
|
|
value = getattr(obj, name)
|
|
if isinstance(value, list):
|
|
# Loop through each item
|
|
dictionary[obj._names[name]] = list()
|
|
for item in value:
|
|
dictionary[obj._names[name]].append(APIHelper.to_dictionary(item) if hasattr(item, "_names") else item)
|
|
elif isinstance(value, dict):
|
|
# Loop through each item
|
|
dictionary[obj._names[name]] = dict()
|
|
for key in value:
|
|
dictionary[obj._names[name]][key] = APIHelper.to_dictionary(value[key]) if hasattr(value[key], "_names") else value[key]
|
|
else:
|
|
dictionary[obj._names[name]] = APIHelper.to_dictionary(value) if hasattr(value, "_names") else value
|
|
|
|
# Return the result
|
|
return dictionary
|
|
|
|
class CustomDate(object):
|
|
|
|
""" A base class for wrapper classes of datetime.
|
|
|
|
This class contains methods which help in
|
|
appropriate serialization of datetime objects.
|
|
|
|
"""
|
|
|
|
def __init__(self, dtime, value=None):
|
|
self.datetime = dtime
|
|
if not value:
|
|
self.value = self.from_datetime(dtime)
|
|
else:
|
|
self.value = value
|
|
|
|
def __str__(self):
|
|
return self.value
|
|
|
|
def __getstate__(self):
|
|
return self.value
|
|
|
|
def __setstate__(self, state):
|
|
pass
|
|
|
|
class HttpDateTime(CustomDate):
|
|
|
|
""" A wrapper class for datetime to support HTTP date format."""
|
|
|
|
@classmethod
|
|
def from_datetime(cls, date_time):
|
|
return eut.formatdate(timeval=mktime(date_time.timetuple()),
|
|
localtime=False, usegmt=True)
|
|
|
|
@classmethod
|
|
def from_value(cls, value):
|
|
dtime = datetime.datetime.fromtimestamp(eut.mktime_tz(eut.parsedate_tz(value)))
|
|
return cls(dtime, value)
|
|
|
|
class UnixDateTime(CustomDate):
|
|
|
|
""" A wrapper class for datetime to support Unix date format."""
|
|
|
|
def __str__(self):
|
|
return str(self.value)
|
|
|
|
@classmethod
|
|
def from_datetime(cls, date_time):
|
|
return calendar.timegm(date_time.utctimetuple())
|
|
|
|
@classmethod
|
|
def from_value(cls, value):
|
|
dtime = datetime.datetime.utcfromtimestamp(float(value))
|
|
return cls(dtime, float(value))
|
|
|
|
class RFC3339DateTime(CustomDate):
|
|
|
|
""" A wrapper class for datetime to support Rfc 3339 format."""
|
|
|
|
@classmethod
|
|
def from_datetime(cls, date_time):
|
|
d = dateutil.parser.parse(date_time).isoformat()
|
|
return d
|
|
|
|
|
|
@classmethod
|
|
def from_value(cls, value):
|
|
dtime = dateutil.parser.parse(value)
|
|
return cls(dtime, value)
|
|
|