""" zeep.wsdl.parse ~~~~~~~~~~~~~~~ """ import typing from lxml import etree from zeep.exceptions import IncompleteMessage, LookupError, NamespaceError from zeep.utils import qname_attr from zeep.wsdl import definitions if typing.TYPE_CHECKING: from zeep.wsdl.wsdl import Definition 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: "Definition", xmlelement: etree._Element ) -> definitions.AbstractMessage: """Create an AbstractMessage object from a xml element. Definition:: * * :param wsdl: The parent definition instance :param xmlelement: The XML node """ tns = wsdl.target_namespace message_name = qname_attr(xmlelement, "name", tns) if not message_name: raise IncompleteMessage("Message element is missing required name attribute") 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) ) message_part = definitions.MessagePart(part_element, part_type) parts.append((part_name, message_part)) # Create the object, add the parts and return it msg = definitions.AbstractMessage(message_name) for part_name, part_value in parts: msg.add_part(part_name, part_value) return msg def parse_abstract_operation( wsdl: "Definition", xmlelement: etree._Element ) -> typing.Optional[definitions.AbstractOperation]: """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 :param xmlelement: The XML node """ name = xmlelement.get("name") kwargs = {"fault_messages": {}} # type: typing.Dict[str, typing.Any] 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") if not param_msg: raise IncompleteMessage( "Operation/%s element is missing required name attribute" % tag_name ) try: param_value = wsdl.get("messages", param_msg.text) except IndexError: return None if tag_name == "input": kwargs["input_message"] = 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")) if wsa_action: kwargs["wsa_action"] = wsa_action elif tag_name == "output": kwargs["output_message"] = param_value else: kwargs["fault_messages"][param_name] = param_value kwargs["name"] = name kwargs["parameter_order"] = xmlelement.get("parameterOrder") return definitions.AbstractOperation(**kwargs) def parse_port_type( wsdl: "Definition", xmlelement: etree._Element ) -> definitions.PortType: """Create a PortType object from a xml element. Definition:: * :param wsdl: The parent definition instance :param xmlelement: The XML node """ name = qname_attr(xmlelement, "name", wsdl.target_namespace) assert name is not None operations = {} # type: typing.Dict[str, definitions.AbstractOperation] 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: "Definition", xmlelement: etree._Element) -> definitions.Port: """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 :param xmlelement: The XML node """ 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: "Definition", xmlelement: etree._Element ) -> definitions.Service: """ Definition:: * ? * ? <-- extensibility element --> <-- extensibility element --> Example:: My first service :param wsdl: The parent definition instance :param xmlelement: The XML node """ name = xmlelement.get("name") ports = [] # type: typing.List[definitions.Port] 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