108 lines
5.0 KiB
Python
108 lines
5.0 KiB
Python
import xml.etree.ElementTree as ET
|
|
from messages import Messages
|
|
|
|
|
|
class RequestXmlService:
|
|
REQUEST_NODE_NAME: str = 'request'
|
|
HEADER_NODE_NAME: str = 'header'
|
|
RESULT_NODE_NAME: str = 'result'
|
|
REQUEST_ID_ATTRIBUTE_NAME: str = 'parcel_id'
|
|
RESPONSE_ID_ATTRIBUTE_NAME: str = 'reply_parcel_id'
|
|
SOURCE_REQUEST_ID_ATTRIBUTE_NAME: str = 'source_parcel_id'
|
|
RESULT_CODE_ATTRIBUTE_NAME: str = 'result_code'
|
|
RESULT_MESSAGE_ATTRIBUTE_NAME: str = 'result_message'
|
|
HEADER_XPATH: str = '/' + REQUEST_NODE_NAME + '/' + HEADER_NODE_NAME
|
|
|
|
def get_request_uuid(self, document: ET.Element) -> str:
|
|
return self.get_header_attribute(document, self.REQUEST_ID_ATTRIBUTE_NAME)
|
|
|
|
def get_response_uuid(self, document: ET.Element) -> str:
|
|
return self.get_header_attribute(document, self.RESPONSE_ID_ATTRIBUTE_NAME)
|
|
|
|
def get_source_request_uuid(self, document: ET.Element) -> str:
|
|
return self.get_header_attribute(document, self.SOURCE_REQUEST_ID_ATTRIBUTE_NAME)
|
|
|
|
def get_header_node(self, document: ET.Element) -> ET.Element:
|
|
return document.find(self.HEADER_XPATH)
|
|
|
|
def get_header_attribute(self, document: ET.Element, attribute_name: str) -> str:
|
|
header = self.get_header_node(document)
|
|
if not header:
|
|
raise Exception(Messages.MISSING_TAG.value)
|
|
result = header.get(attribute_name)
|
|
return result.strip() if result else ''
|
|
|
|
def get_request_document(self, request_uuid: str, response_uuid: str | None) -> ET.Element:
|
|
document = ET.Element(self.REQUEST_NODE_NAME)
|
|
header = ET.SubElement(document, self.HEADER_NODE_NAME)
|
|
header.set(self.REQUEST_ID_ATTRIBUTE_NAME, request_uuid.strip())
|
|
if response_uuid:
|
|
header.set(self.RESPONSE_ID_ATTRIBUTE_NAME, response_uuid.strip())
|
|
return document
|
|
|
|
def get_common_response_xml(self, request_uuid: str, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
|
|
document = self.get_request_document(response_uuid, request_uuid)
|
|
|
|
result = ET.SubElement(document, self.RESULT_NODE_NAME)
|
|
result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
|
|
result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
|
|
|
|
return document
|
|
|
|
def set_result(self, document: ET.Element, result_code: int, result_message: str):
|
|
result = document.find(self.RESULT_NODE_NAME)
|
|
result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
|
|
result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
|
|
|
|
def set_request_uuid(self, document: ET.Element, request_uuid: str):
|
|
header = self.get_header_node(document)
|
|
header.set(self.REQUEST_ID_ATTRIBUTE_NAME, request_uuid.strip())
|
|
|
|
def set_source_request_uuid(self, document: ET.Element, source_request_uuid: str):
|
|
header = self.get_header_node(document)
|
|
header.set(self.SOURCE_REQUEST_ID_ATTRIBUTE_NAME, source_request_uuid.strip())
|
|
|
|
def get_common_request_xml(self, request_uuid: str) -> ET.Element:
|
|
return self.get_request_document(request_uuid, None)
|
|
|
|
def get_response_xml(self, document: ET.Element, request_uuid: str | None, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
|
|
header = document.find(self.HEADER_NODE_NAME)
|
|
header.set(self.REQUEST_ID_ATTRIBUTE_NAME, response_uuid.strip())
|
|
header.set(self.RESPONSE_ID_ATTRIBUTE_NAME, request_uuid.strip() if request_uuid else None)
|
|
result = document.find(self.RESULT_NODE_NAME)
|
|
result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
|
|
result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
|
|
|
|
def get_element(self, document: ET.Element, node: ET.Element, name: str) -> ET.Element:
|
|
for child in node:
|
|
if child.tag.casefold() == name.casefold():
|
|
return child
|
|
return ET.SubElement(document, name)
|
|
|
|
def validate_common_request_xml(self, xml: str):
|
|
if not xml:
|
|
raise Exception(Messages.NO_REQUEST_DATA_PROVIDED.name)
|
|
try:
|
|
document = ET.parse(xml)
|
|
except:
|
|
raise Exception(Messages.INVALID_REQUEST_DATA)
|
|
self.validate_common_request_xml(document)
|
|
|
|
def validate_common_request_xml(self, document: ET.Element | None):
|
|
if not document:
|
|
raise Exception(Messages.NO_REQUEST_DATA_PROVIDED.name)
|
|
if document.tag != self.REQUEST_NODE_NAME:
|
|
raise Exception(Messages.MISSING_TAG.name, self.REQUEST_NODE_NAME)
|
|
try:
|
|
request_uuid = self.get_header_attribute(document, self.REQUEST_ID_ATTRIBUTE_NAME)
|
|
except:
|
|
raise Exception(Messages.INVALID_REQUEST_DATA)
|
|
if not request_uuid:
|
|
raise Exception(Messages.MISSING_ATTRIBUTE, self.REQUEST_ID_ATTRIBUTE_NAME, self.HEADER_NODE_NAME)
|
|
|
|
def get_result_code(self, document: ET.Element) -> str | None:
|
|
result = self.get_element(document, document, self.RESULT_NODE_NAME)
|
|
if not result:
|
|
return None
|
|
return result.get(self.RESULT_CODE_ATTRIBUTE_NAME)
|