Created
October 23, 2019 11:27
-
-
Save nicoknoll/7496874986fd208aa42c8db2f7c8604b to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class CxmlNode: | |
dtd = etree.DTD(dtd_file) | |
def __init__(self, *args, **kwargs): | |
self.payload_id = kwargs['payload_id'] | |
self.timestamp = kwargs.get('timestamp') # datetime.now().isoformat() | |
def __str__(self): | |
return self.render().decode('UTF-8') | |
@cached_property | |
def tree(self): | |
return self.get_root() | |
def get_root(self): | |
root = Element("cXML", payloadID=self.payload_id, timestamp=self.timestamp) | |
root.attrib['{http://www.w3.org/XML/1998/namespace}lang'] = "en" | |
return root | |
def populate_tree(self): | |
pass | |
def render(self): | |
self.populate_tree() | |
return etree.tostring( | |
self.tree, | |
pretty_print=True, | |
xml_declaration=True, | |
encoding='UTF-8', | |
doctype=( | |
'<!DOCTYPE cXML SYSTEM ' | |
'"http://xml.cXML.org/schemas/cXML/1.2.014/Fulfill.dtd">' | |
) | |
) | |
@classmethod | |
def validate_xml(cls, string): | |
tree = etree.fromstring(string) | |
status = cls.dtd.validate(tree) # status is a Boolean | |
if status is False: | |
logger.error(cls.dtd.error_log.filter_from_errors()) | |
return status | |
@classmethod | |
def get_kwargs_from_xml(cls, string): | |
xml = etree.fromstring(string) | |
return { | |
'payload_id': xml.attrib["payloadID"], | |
'timestamp': xml.attrib["timestamp"], | |
} | |
@classmethod | |
def create_from_xml(cls, string): | |
return cls(**cls.get_kwargs_from_xml(string)) | |
class Response(CxmlNode): | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) | |
self.status_code = kwargs.get('status_code') | |
self.status_text = kwargs.get('status_text') | |
def populate_tree(self): | |
resp = SubElement(self.tree, "Response") | |
SubElement(resp, 'Status', code=self.status_code, text=self.status_text) | |
class Request(CxmlNode): | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) | |
self.from_domain = kwargs.get('from_domain') | |
self.from_identity = kwargs.get('from_identity') | |
self.to_domain = kwargs.get('to_domain') | |
self.to_identity = kwargs.get('to_identity') | |
self.sender_domain = kwargs.get('sender_domain') | |
self.sender_identity = kwargs.get('sender_identity') | |
self.sender_shared_secret = kwargs.get('sender_shared_secret') | |
self.sender_user_agent = kwargs.get('sender_user_agent') | |
self.deployment_mode = kwargs.get('deployment_mode') | |
def _populate_header(self): | |
header = SubElement(self.tree, 'Header') | |
header_from = SubElement(header, 'From') | |
cred = SubElement(header_from, 'Credential', domain=self.from_domain) | |
identity = SubElement(cred, 'Identity') | |
identity.text = self.from_identity | |
header_to = SubElement(header, 'To') | |
cred = SubElement(header_to, 'Credential', domain=self.to_domain) | |
identity = SubElement(cred, 'Identity') | |
identity.text = self.to_identity | |
header_sender = SubElement(header, 'Sender') | |
cred = SubElement(header_sender, 'Credential', domain=self.sender_domain) | |
identity = SubElement(cred, 'Identity') | |
identity.text = self.sender_identity | |
secret = SubElement(cred, 'SharedSecret') | |
secret.text = self.sender_shared_secret | |
ua = SubElement(header_sender, 'UserAgent') | |
ua.text = self.sender_user_agent | |
return header | |
def _populate_request(self): | |
return SubElement(self.tree, 'Request', deploymentMode=self.deployment_mode) | |
def populate_tree(self): | |
self._header = self._populate_header() | |
self._request = self._populate_request() | |
@classmethod | |
def get_kwargs_from_xml(cls, string): | |
xml = etree.fromstring(string) | |
kwargs = super().get_kwargs_from_xml(string) | |
kwargs.update({ | |
'from_domain': xml.find("Header/From/Credential").attrib["domain"], | |
'from_identity': get_text(xml.find("Header/From/Credential/Identity")), | |
'to_domain': xml.find("Header/To/Credential").attrib["domain"], | |
'to_identity': get_text(xml.find("Header/To/Credential/Identity")), | |
'sender_domain': xml.find("Header/Sender/Credential").attrib["domain"], | |
'sender_identity': get_text(xml.find("Header/Sender/Credential/Identity")), | |
'sender_shared_secret': get_text(xml.find("Header/Sender/Credential/SharedSecret")), | |
'sender_user_agent': get_text(xml.find("Header/Sender/UserAgent")), | |
'deployment_mode': xml.find("Request").attrib["deploymentMode"], | |
}) | |
return kwargs |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment