Source code for cloudshell.api.common_cloudshell_api

# -*- coding: utf-8 -*-

import importlib
import types
import ssl
import sys
import urllib3
import urllib3.util as urllib_util
import xml.etree.ElementTree as etree

from collections import OrderedDict
from xml.sax.saxutils import escape


if sys.version_info.major == 2:
    unicode = unicode
    str = str
    bytes = str
    basestring = basestring
    TYPE_TYPE = types.TypeType
    TYPE_CLASS = types.ClassType
elif sys.version_info.major == 3:
    str = str
    unicode = str
    bytes = bytes
    basestring = (str, bytes)
    TYPE_TYPE = type
    TYPE_CLASS = type

[docs]class XMLWrapper:
[docs] @staticmethod def parseXML(xml_str): return etree.fromstring(xml_str)
[docs] @staticmethod def getRootNode(node): return node.getroot()
[docs] @staticmethod def getChildNode(parent_node, child_name, find_prefix=''): return parent_node.find(find_prefix + child_name)
[docs] @staticmethod def getAllChildNode(parent_node, child_name, find_prefix=''): return parent_node.findall(find_prefix + child_name)
[docs] @staticmethod def getChildNodeByAttr(parent_node, child_name, attr_name, attr_value): return parent_node.find(child_name + '[@' + attr_name + '=\'' + attr_value + '\']')
[docs] @staticmethod def getAllChildNodeByAttr(parent_node, child_name, attr_name, attr_value): return parent_node.findall(child_name + '[@' + attr_name + '=\'' + attr_value + '\']')
[docs] @staticmethod def getNodeName(node): return node.tag
[docs] @staticmethod def getNodeText(node): return node.text
[docs] @staticmethod def getNodeAttr(node, attribute_name, find_prefix=''): return node.get(find_prefix + attribute_name)
[docs] @staticmethod def getNodePrefix(node, prefix_name): prefix = '' if len(node.attrib) == 0: return prefix for attrib_name, value in node.attrib.items(): if attrib_name[0] == "{": prefix, ignore, tag = attrib_name[1:].partition("}") return "{" + prefix + "}" return prefix
[docs] @staticmethod def getStringFromXML(node, pretty_print=False): return etree.tostring(node, pretty_print=pretty_print)
# map request class
[docs]class CommonAPIRequest: def __init__(self, **kwarg): self.attributes = [] for key, value in sorted(kwarg.items()): self.attributes.append(key) setattr(self, key, value) @staticmethod def _checkContainerValue(value): result_value = None if isinstance(value, list): result_value = list() for list_value in value: result_value.append(CommonAPIRequest.toContainer(list_value)) elif isinstance(value, CommonAPIRequest): result_value = CommonAPIRequest.toContainer(value) else: result_value = value return result_value
[docs] @staticmethod def toContainer(data): if isinstance(data, dict) or isinstance(data, OrderedDict): return data if isinstance(data, list): data_list = list() for value in data: data_list.append(CommonAPIRequest._checkContainerValue(value)) return data_list data_dict = OrderedDict() data_dict['__name__'] = data.__class__.__name__ for key in data.attributes: data_dict[key] = CommonAPIRequest._checkContainerValue(getattr(data, key)) # for key, value in data.__dict__.items(): # data_dict[key] = CommonAPIRequest._checkContainerValue(value) return data_dict
# end map request class
[docs]class CommonResponseInfo: def __init__(self, xml_object, find_prefix): self._parseAttributesData(self.__class__, xml_object, find_prefix) def _attributeCastToType(self, data_str, cast_type_name): default_value = 0 if cast_type_name == 'bool': default_value = False elif cast_type_name == 'float': default_value = 0.0 elif cast_type_name == 'str': default_value = '' cast_type = eval(cast_type_name) data = None if data_str is not None: data = default_value try: if cast_type_name == 'bool': data = (data_str.lower() in ['true', '1', 'yes', 'on']) else: data = cast_type(data_str) except UnicodeEncodeError as err: try: data = data_str.encode('utf-8') except: pass except ValueError as err: pass return data def _isAttributeTypeDefault(self, attr_type_name): return (attr_type_name == 'int' or attr_type_name == 'long' or attr_type_name == 'float' or attr_type_name == 'bool' or attr_type_name == 'str') def _is_empty_object(self, atrrib_data): for key, value in atrrib_data.items(): if isinstance(value, list) and len(value) > 0: return False if value is not None: return False return True def _append_object_list(self, attr_type_name, list_node, attr_type_instance, class_type, find_prefix): if self._isAttributeTypeDefault(attr_type_name): data_str = XMLWrapper.getNodeText(list_node) data = self._attributeCastToType(data_str, attr_type_name) else: if attr_type_instance == object: data = class_type(list_node, find_prefix) else: data = attr_type_instance(list_node, find_prefix) if not (hasattr(list_node, "attrib") and list_node.attrib): setattr(data, "is_empty_object", True) if hasattr(data, "is_empty_object") and data.is_empty_object: return None else: return data def _parseAttributesData(self, class_type, xml_object, find_prefix): attrib_data_dict = dict() empty_object_size = len(self.__dict__) for name, attr_type in self.__dict__.items(): if not isinstance(attr_type, (TYPE_TYPE, TYPE_CLASS)) and not isinstance(attr_type, dict): continue if not isinstance(attr_type, dict): data = None attr_type_name = attr_type.__name__ if self._isAttributeTypeDefault(attr_type_name): data_str = XMLWrapper.getNodeAttr(xml_object, name) if data_str is None: child_attribute = XMLWrapper.getChildNode(xml_object, name) if child_attribute is not None: data_str = XMLWrapper.getNodeText(child_attribute) data = self._attributeCastToType(data_str, attr_type_name) else: child_node = XMLWrapper.getChildNode(xml_object, name) if child_node is not None: child_type = XMLWrapper.getNodeAttr(child_node, 'type', find_prefix) if child_type is None: data = attr_type(child_node, find_prefix) else: data = child_type(child_node, find_prefix) else: # continue data = None attrib_data_dict[name] = data else: child_node = XMLWrapper.getChildNode(xml_object, name) data_list = list() attr_type_instance = attr_type['list'] attr_type_name = attr_type_instance.__name__ if child_node is not None: child_count = 0 for list_node in child_node: data_object = self._append_object_list(attr_type_name, list_node, attr_type_instance, class_type, find_prefix) if data_object is not None: data_list.append(data_object) child_count += 1 # I think that it is a logical bug, but ... if child_count == 0: for list_node in xml_object: if XMLWrapper.getNodeName(list_node) == name: data_object = self._append_object_list(attr_type_name, list_node, attr_type_instance, class_type, find_prefix) if data_object is not None: data_list.append(data_object) attrib_data_dict[name] = data_list if not self._is_empty_object(attrib_data_dict): for key, value in attrib_data_dict.items(): setattr(self, key, value) elif len(self.__dict__) == empty_object_size: setattr(self, "is_empty_object", True)
[docs]class CommonApiResult: def __init__(self, xml_object): error_node = XMLWrapper.getChildNode(xml_object, 'Error') self.error = None if error_node is None else XMLWrapper.getNodeText(error_node) error_code_node = XMLWrapper.getChildNode(xml_object, 'ErrorCode') self.error_code = None if error_code_node is None else XMLWrapper.getNodeText(error_code_node) self.response_info = None response_info_node = XMLWrapper.getChildNode(xml_object, 'ResponseInfo') if response_info_node is not None: find_prefix = XMLWrapper.getNodePrefix(response_info_node, 'xsi') type_attr = XMLWrapper.getNodeAttr(response_info_node, find_prefix + 'type') if type_attr is not None: response_class = CommonApiResult.importAPIClass(type_attr) if response_class is not None: self.response_info = response_class(response_info_node, find_prefix) success = XMLWrapper.getNodeAttr(xml_object, 'Success') success = success.lower() self.success = success in ['true', 'yes', 'on']
[docs] @staticmethod def importAPIClass(name): module = importlib.import_module('cloudshell.api.cloudshell_api') if hasattr(module, name): return getattr(module, name) return None
[docs]class CloudShellAPIError(Exception): def __init__(self, code, message, rawxml): self.code = code self.message = message self.rawxml = rawxml def __str__(self): return 'CloudShell API error ' + str(self.code) + ': ' + self.message def __repr__(self): return 'CloudShell API error ' + str(self.code) + ': ' + self.message
[docs]class CommonAPISession: def __init__(self, host, username, password, domain, connection_pool_size=25): = host self.username = username self.password = password self.domain = domain if sys.version_info[0] == 2 and sys.version_info[2] < 13: ssl_protocol = ssl.PROTOCOL_SSLv23 else: ssl_protocol = ssl.PROTOCOL_TLS ctx = urllib_util.SSLContext(ssl_protocol) ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE self._http = urllib3.PoolManager(num_pools=connection_pool_size, ssl_context=ctx) def _parseXML(self, xml_str): return etree.fromstring(xml_str) def _replaceSendValue(self, data): """Normalize xml string, escape special xml characters """ if data is None: return u'' try: data_str = unicode(data) except: data_str = unicode(data.decode("utf-8")) data_str = u"".join([escape(char) for char in data_str]) if data_str == 'True' or data_str == 'False': return data_str.lower() else: return data_str def _to_unicode_string(self, data): if data is None: return u'' try: return unicode(data) except: return unicode(data.decode("utf-8")) def _sendRequest(self, operation, message, request_headers): """ Sending http POST request through URLLIB package :param operation: operation name :param message: request body :param request_headers: header of the request :return: responce string data """ operation_url = str(self.url + operation) response = self._http.request("POST", operation_url, body=message.encode('utf-8'), headers=request_headers) if not raise Exception(response.reason) return def _new_serializeRequestData(self, root_node, object_data, prev_type=None): """Generate xml from received request data using etree.xml """ if isinstance(object_data, dict): if '__name__' in object_data: working_node = etree.SubElement(root_node, object_data.pop('__name__')) else: working_node = root_node for key, value in object_data.items(): if value is None: continue if isinstance(value, basestring): new_node = etree.SubElement(working_node, key) new_node.text = value elif isinstance(value, bool): new_node = etree.SubElement(working_node, key) new_node.text = str(value).lower() else: child_node = working_node if isinstance(value, list): child_node = etree.SubElement(working_node, key) serialized_node = self._new_serializeRequestData(child_node, value) return root_node elif isinstance(object_data, list): for value in object_data: serialized_node = self._new_serializeRequestData(root_node, value, list()) elif isinstance(object_data, basestring) or isinstance(object_data, int) or isinstance(object_data, float): if prev_type is not None and isinstance(prev_type, list): child_node = etree.SubElement(root_node, 'string') child_node.text = object_data elif isinstance(object_data, bool): root_node.text = str(object_data).lower() else: root_node.text = self._to_unicode_string(object_data) return root_node
[docs] def generateAPIRequest(self, kwargs): """ Generic method for generation and sending XML requests :param return_type: type of returning data :param kwargs: map of the parameters that need to be send to the server :return: string data or API object """ if 'method_name' not in kwargs: raise CloudShellAPIError(404, 'Key "method_name" not in input data!', '') method_name = kwargs.pop('method_name', None) message = self._serialize_request(kwargs, method_name) result = self._handle_api_response(self._sendRequest(method_name, message)) return result
def _handle_api_response(self, response_str): ''' :param str response_str: :return: either string or deserialized instance representing the api response ''' response_str = self._remove_xml_namespace(response_str) try: api_response = self._deserialize_response(response_str) except Exception: raise Exception("Invalid Request: " + str(response_str)) if not api_response.success: raise CloudShellAPIError(api_response.error_code, api_response.error, response_str) result = response_str if api_response.response_info: result = api_response.response_info return result def _serialize_request(self, kwargs, method_name): request_node = etree.Element(method_name) # request_str = '<' + method_name + '>\n' for name in kwargs: child_node = etree.SubElement(request_node, name) self._new_serializeRequestData(child_node, kwargs[name]) return etree.tostring(request_node).decode("utf-8") def _deserialize_response(self, response_str): response_xml = XMLWrapper.parseXML(response_str) api_result = CommonApiResult(response_xml) return api_result def _remove_xml_namespace(self, response_str): return response_str.replace(b'xmlns=""', b'') \ .replace(b'&#x0;', b'<NUL>') def __prettify_xml(self, elem): """Return a pretty-printed XML string for the Element. """ from xml.dom.minidom import parseString rough_string = etree.tostring(elem, 'utf-8') reparsed = parseString(rough_string) return reparsed.toprettyxml(indent="\t")