Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_or_create_service_node(self, service_name):
"""Builds a wsdl:service element."""
ser = None
if not service_name in self.service_elt_dict:
ser = SubElement(self.root_elt, WSDL11("service"))
ser.set('name', service_name)
self.service_elt_dict[service_name] = ser
else:
ser = self.service_elt_dict[service_name]
return ser
binding_name = self._get_binding_name(port_type_name)
port_binding_names.append((port_type_name, binding_name))
else:
port_type = self._get_or_create_port_type(service_name)
port_type.set('name', service_name)
binding_name = self._get_binding_name(service_name)
port_binding_names.append((service_name, binding_name))
for method in service.public_methods.values():
check_method_port(service, method)
if method.is_callback:
operation = SubElement(cb_port_type, WSDL11("operation"))
else:
operation = SubElement(port_type, WSDL11("operation"))
operation.set('name', method.operation_name)
if method.doc is not None:
operation.append(E(WSDL11("documentation"), method.doc))
operation.set('parameterOrder', method.in_message.get_element_name())
op_input = SubElement(operation, WSDL11("input"))
op_input.set('name', method.in_message.get_element_name())
op_input.set('message',
method.in_message.get_element_name_ns(self.interface))
if (not method.is_callback) and (not method.is_async):
if len(in_headers) > 1:
in_header_message_name = ''.join((method.name,
_in_header_msg_suffix))
else:
in_header_message_name = in_headers[0].get_type_name()
for header in in_headers:
soap_header = SubElement(input, WSDL11_SOAP('header'))
soap_header.set('use', 'literal')
soap_header.set('message', '%s:%s' % (
header.get_namespace_prefix(self.interface),
in_header_message_name))
soap_header.set('part', header.get_type_name())
if not (method.is_async or method.is_callback):
output = SubElement(operation, WSDL11("output"))
output.set('name', method.out_message.get_element_name())
soap_body = SubElement(output, WSDL11_SOAP("body"))
soap_body.set('use', 'literal')
# get output soap header
out_header = method.out_header
if out_header is None:
out_header = service.__out_header__
if not (out_header is None):
if isinstance(out_header, (list, tuple)):
out_headers = out_header
else:
out_headers = (out_header,)
reply_part.set('name', 'ReplyTo')
reply_part.set('element', '%s:ReplyTo' % PREF_WSA)
id_header = SubElement(root, WSDL11("message"))
id_header.set('name', 'MessageIDHeader')
id_part = SubElement(id_header, WSDL11("part"))
id_part.set('name', 'MessageID')
id_part.set('element', '%s:MessageID' % PREF_WSA)
# make portTypes
cb_port_type = SubElement(root, WSDL11("portType"))
cb_port_type.set('name', '%sCallback' % service_name)
cb_service_name = '%sCallback' % service_name
cb_service = SubElement(root, WSDL11("service"))
cb_service.set('name', cb_service_name)
cb_wsdl_port = SubElement(cb_service, WSDL11("port"))
cb_wsdl_port.set('name', cb_service_name)
cb_wsdl_port.set('binding', '%s:%s' % (pref_tns, cb_service_name))
cb_address = SubElement(cb_wsdl_port, WSDL11_SOAP("address"))
cb_address.set('location', url)
return cb_port_type
import_.set("namespace", NS_WSA)
import_.set("schemaLocation", NS_WSA)
relt_message = SubElement(root, WSDL11("message"))
relt_message.set('name', 'RelatesToHeader')
relt_part = SubElement(relt_message, WSDL11("part"))
relt_part.set('name', 'RelatesTo')
relt_part.set('element', '%s:RelatesTo' % PREF_WSA)
reply_message = SubElement(root, WSDL11("message"))
reply_message.set('name', 'ReplyToHeader')
reply_part = SubElement(reply_message, WSDL11("part"))
reply_part.set('name', 'ReplyTo')
reply_part.set('element', '%s:ReplyTo' % PREF_WSA)
id_header = SubElement(root, WSDL11("message"))
id_header.set('name', 'MessageIDHeader')
id_part = SubElement(id_header, WSDL11("part"))
id_part.set('name', 'MessageID')
id_part.set('element', '%s:MessageID' % PREF_WSA)
# make portTypes
cb_port_type = SubElement(root, WSDL11("portType"))
cb_port_type.set('name', '%sCallback' % service_name)
cb_service_name = '%sCallback' % service_name
cb_service = SubElement(root, WSDL11("service"))
cb_service.set('name', cb_service_name)
cb_wsdl_port = SubElement(cb_service, WSDL11("port"))
cb_wsdl_port.set('name', cb_service_name)
def inner(method, binding):
operation = etree.Element(WSDL11("operation"))
operation.set('name', method.operation_name)
soap_operation = SubElement(operation, WSDL11_SOAP("operation"))
soap_operation.set('soapAction', method.operation_name)
soap_operation.set('style', 'document')
# get input
input = SubElement(operation, WSDL11("input"))
input.set('name', method.in_message.get_element_name())
soap_body = SubElement(input, WSDL11_SOAP("body"))
soap_body.set('use', 'literal')
# get input soap header
in_header = method.in_header
if in_header is None:
# add necessary async headers
# WS-Addressing -> RelatesTo ReplyTo MessageID
# callback porttype
if service._has_callbacks():
wsa_schema = SubElement(types, XSD("schema"))
wsa_schema.set("targetNamespace", '%sCallback' % ns_tns)
wsa_schema.set("elementFormDefault", "qualified")
import_ = SubElement(wsa_schema, XSD("import"))
import_.set("namespace", NS_WSA)
import_.set("schemaLocation", NS_WSA)
relt_message = SubElement(root, WSDL11("message"))
relt_message.set('name', 'RelatesToHeader')
relt_part = SubElement(relt_message, WSDL11("part"))
relt_part.set('name', 'RelatesTo')
relt_part.set('element', '%s:RelatesTo' % PREF_WSA)
reply_message = SubElement(root, WSDL11("message"))
reply_message.set('name', 'ReplyToHeader')
reply_part = SubElement(reply_message, WSDL11("part"))
reply_part.set('name', 'ReplyTo')
reply_part.set('element', '%s:ReplyTo' % PREF_WSA)
id_header = SubElement(root, WSDL11("message"))
id_header.set('name', 'MessageIDHeader')
id_part = SubElement(id_header, WSDL11("part"))
id_part.set('name', 'MessageID')
id_part.set('element', '%s:MessageID' % PREF_WSA)
# make portTypes