""" zeep.wsdl.parse ~~~~~~~~~~~~~~~ """ from lxml import etree from zeep.exceptions import IncompleteMessage, LookupError, NamespaceError from zeep.utils import qname_attr from zeep.wsdl import definitions NSMAP = { "wsdl": "http://schemas.xmlsoap.org/wsdl/", "wsaw": "http://www.w3.org/2006/05/addressing/wsdl", "wsam": "http://www.w3.org/2007/05/addressing/metadata", } def parse_abstract_message(wsdl, xmlelement): """Create an AbstractMessage object from a xml element. Definition:: * * :param wsdl: The parent definition instance :type wsdl: zeep.wsdl.wsdl.Definition :param xmlelement: The XML node :type xmlelement: lxml.etree._Element :rtype: zeep.wsdl.definitions.AbstractMessage """ tns = wsdl.target_namespace message_name = qname_attr(xmlelement, "name", tns) parts = [] for part in xmlelement.findall("wsdl:part", namespaces=NSMAP): part_name = part.get("name") part_element = qname_attr(part, "element") part_type = qname_attr(part, "type") try: if part_element is not None: part_element = wsdl.types.get_element(part_element) if part_type is not None: part_type = wsdl.types.get_type(part_type) except (NamespaceError, LookupError): raise IncompleteMessage( ( "The wsdl:message for %r contains an invalid part (%r): " "invalid xsd type or elements" ) % (message_name.text, part_name) ) part = definitions.MessagePart(part_element, part_type) parts.append((part_name, part)) # Create the object, add the parts and return it msg = definitions.AbstractMessage(message_name) for part_name, part in parts: msg.add_part(part_name, part) return msg def parse_abstract_operation(wsdl, xmlelement): """Create an AbstractOperation object from a xml element. This is called from the parse_port_type function since the abstract operations are part of the port type element. Definition:: * ? ? ? ? ? * ? :param wsdl: The parent definition instance :type wsdl: zeep.wsdl.wsdl.Definition :param xmlelement: The XML node :type xmlelement: lxml.etree._Element :rtype: zeep.wsdl.definitions.AbstractOperation """ name = xmlelement.get("name") kwargs = {"fault_messages": {}} for msg_node in xmlelement: tag_name = etree.QName(msg_node.tag).localname if tag_name not in ("input", "output", "fault"): continue param_msg = qname_attr(msg_node, "message", wsdl.target_namespace) param_name = msg_node.get("name") try: param_value = wsdl.get("messages", param_msg.text) except IndexError: return if tag_name == "input": kwargs["input_message"] = param_value elif tag_name == "output": kwargs["output_message"] = param_value else: kwargs["fault_messages"][param_name] = param_value wsa_action = msg_node.get(etree.QName(NSMAP["wsam"], "Action")) if not wsa_action: wsa_action = msg_node.get(etree.QName(NSMAP["wsaw"], "Action")) param_value.wsa_action = wsa_action kwargs["name"] = name kwargs["parameter_order"] = xmlelement.get("parameterOrder") return definitions.AbstractOperation(**kwargs) def parse_port_type(wsdl, xmlelement): """Create a PortType object from a xml element. Definition:: * :param wsdl: The parent definition instance :type wsdl: zeep.wsdl.wsdl.Definition :param xmlelement: The XML node :type xmlelement: lxml.etree._Element :rtype: zeep.wsdl.definitions.PortType """ name = qname_attr(xmlelement, "name", wsdl.target_namespace) operations = {} for elm in xmlelement.findall("wsdl:operation", namespaces=NSMAP): operation = parse_abstract_operation(wsdl, elm) if operation: operations[operation.name] = operation return definitions.PortType(name, operations) def parse_port(wsdl, xmlelement): """Create a Port object from a xml element. This is called via the parse_service function since ports are part of the service xml elements. Definition:: * ? <-- extensibility element --> :param wsdl: The parent definition instance :type wsdl: zeep.wsdl.wsdl.Definition :param xmlelement: The XML node :type xmlelement: lxml.etree._Element :rtype: zeep.wsdl.definitions.Port """ name = xmlelement.get("name") binding_name = qname_attr(xmlelement, "binding", wsdl.target_namespace) return definitions.Port(name, binding_name=binding_name, xmlelement=xmlelement) def parse_service(wsdl, xmlelement): """ Definition:: * ? * ? <-- extensibility element --> <-- extensibility element --> Example:: My first service :param wsdl: The parent definition instance :type wsdl: zeep.wsdl.wsdl.Definition :param xmlelement: The XML node :type xmlelement: lxml.etree._Element :rtype: zeep.wsdl.definitions.Service """ name = xmlelement.get("name") ports = [] for port_node in xmlelement.findall("wsdl:port", namespaces=NSMAP): port = parse_port(wsdl, port_node) if port: ports.append(port) obj = definitions.Service(name) for port in ports: obj.add_port(port) return obj