<s> from collections import OrderedDict , namedtuple <EOL> from lxml import etree <EOL> from zeep . utils import qname_attr <EOL> NSMAP = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> MessagePart = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:type>' ] ) <EOL> class AbstractMessage ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> self . parts = OrderedDict ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . name . text ) <EOL> def resolve ( self , definitions ) : <EOL> pass <EOL> def add_part ( self , name , element ) : <EOL> self . parts [ name ] = element <EOL> def get_part ( self , name ) : <EOL> return self . parts [ name ] <EOL> @ classmethod <EOL> def parse ( cls , definitions , xmlelement ) : <EOL> """<STR_LIT>""" <EOL> msg = cls ( name = qname_attr ( xmlelement , '<STR_LIT:name>' , definitions . target_namespace ) ) <EOL> for part in xmlelement . findall ( '<STR_LIT>' , namespaces = NSMAP ) : <EOL> part_name = part . get ( '<STR_LIT:name>' ) <EOL> part_element = qname_attr ( part , '<STR_LIT>' , definitions . target_namespace ) <EOL> part_type = qname_attr ( part , '<STR_LIT:type>' , definitions . target_namespace ) <EOL> if part_element is not None : <EOL> part_element = definitions . schema . get_element ( part_element ) <EOL> if part_type is not None : <EOL> part_type = definitions . schema . get_type ( part_type ) <EOL> msg . add_part ( part_name , MessagePart ( part_element , part_type ) ) <EOL> return msg <EOL> class AbstractOperation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , input = None , output = None , faults = None , <EOL> parameter_order = None ) : <EOL> self . name = name <EOL> self . input = input <EOL> self . output = output <EOL> self . faults = faults <EOL> self . parameter_order = parameter_order <EOL> def get ( self , type_ , name = None ) : <EOL> if type_ in ( '<STR_LIT:input>' , '<STR_LIT>' ) : <EOL> return getattr ( self , type_ ) <EOL> return self . faults [ name ] <EOL> @ classmethod <EOL> def parse ( cls , definitions , xmlelement ) : <EOL> """<STR_LIT>""" <EOL> name = xmlelement . get ( '<STR_LIT:name>' ) <EOL> kwargs = { <EOL> '<STR_LIT>' : { } <EOL> } <EOL> for msg_node in xmlelement . getchildren ( ) : <EOL> tag_name = etree . QName ( msg_node . tag ) . localname <EOL> if tag_name not in ( '<STR_LIT:input>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> continue <EOL> param_msg = qname_attr ( msg_node , '<STR_LIT:message>' , definitions . target_namespace ) <EOL> param_name = msg_node . get ( '<STR_LIT:name>' ) <EOL> if tag_name in ( '<STR_LIT:input>' , '<STR_LIT>' ) : <EOL> kwargs [ tag_name ] = definitions . messages [ param_msg . text ] <EOL> else : <EOL> kwargs [ '<STR_LIT>' ] [ param_name ] = definitions . messages [ param_msg . text ] <EOL> kwargs [ '<STR_LIT:name>' ] = name <EOL> kwargs [ '<STR_LIT>' ] = xmlelement . get ( '<STR_LIT>' ) <EOL> return cls ( ** kwargs ) <EOL> class PortType ( object ) : <EOL> def __init__ ( self , name , operations ) : <EOL> self . name = name <EOL> self . operations = operations <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . name . text ) <EOL> def resolve ( self , definitions ) : <EOL> pass <EOL> @ classmethod <EOL> def parse ( cls , definitions , xmlelement ) : <EOL> """<STR_LIT>""" <EOL> name = qname_attr ( xmlelement , '<STR_LIT:name>' , definitions . target_namespace ) <EOL> operations = { } <EOL> for elm in xmlelement . findall ( '<STR_LIT>' , namespaces = NSMAP ) : <EOL> operation = AbstractOperation . parse ( definitions , elm ) <EOL> operations [ operation . name ] = operation <EOL> return cls ( name , operations ) <EOL> class Binding ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , wsdl , name , port_name ) : <EOL> self . name = name <EOL> self . port_name = port_name <EOL> self . port_type = None <EOL> self . wsdl = wsdl <EOL> self . _operations = { } <EOL> def resolve ( self , definitions ) : <EOL> self . port_type = definitions . port_types [ self . port_name . text ] <EOL> for operation in self . _operations . values ( ) : <EOL> operation . resolve ( definitions ) <EOL> def _operation_add ( self , operation ) : <EOL> self . _operations [ operation . name ] = operation <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . name . text , self . port_type ) <EOL> def get ( self , name ) : <EOL> return self . _operations . get ( name ) <EOL> @ classmethod <EOL> def match ( cls , node ) : <EOL> raise NotImplementedError ( ) <EOL> @ classmethod <EOL> def parse ( cls , definitions , xmlelement ) : <EOL> raise NotImplementedError ( ) <EOL> class ConcreteMessage ( object ) : <EOL> def __init__ ( self , wsdl , name , operation ) : <EOL> self . wsdl = wsdl <EOL> self . namespace = { } <EOL> self . operation = operation <EOL> self . name = name <EOL> def create ( self , * args , ** kwargs ) : <EOL> raise NotImplementedError ( ) <EOL> @ classmethod <EOL> def parse ( cls , wsdl , xmlelement , abstract_message , operation ) : <EOL> raise NotImplementedError ( ) <EOL> def signature ( self , as_output = False ) : <EOL> if as_output : <EOL> return self . body . type . name <EOL> return self . body . type . signature ( ) <EOL> class Operation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , binding ) : <EOL> self . name = name <EOL> self . binding = binding <EOL> self . abstract = None <EOL> self . style = None <EOL> self . input = None <EOL> self . output = None <EOL> self . faults = { } <EOL> def resolve ( self , definitions ) : <EOL> self . abstract = self . binding . port_type . operations [ self . name ] <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . name , self . style ) <EOL> def __unicode__ ( self ) : <EOL> if not self . input : <EOL> return '<STR_LIT>' % ( self . name ) <EOL> retval = '<STR_LIT>' % ( self . name , self . input . signature ( ) ) <EOL> if self . output : <EOL> retval += '<STR_LIT>' % ( self . output . signature ( as_output = True ) ) <EOL> return retval <EOL> def create ( self , * args , ** kwargs ) : <EOL> return self . input . serialize ( * args , ** kwargs ) <EOL> def process_reply ( self , envelope ) : <EOL> raise NotImplementedError ( ) <EOL> @ classmethod <EOL> def parse ( cls , wsdl , xmlelement , binding ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> class Port ( object ) : <EOL> def __init__ ( self , name , binding_name , xmlelement ) : <EOL> self . name = name <EOL> self . _resolve_context = { <EOL> '<STR_LIT>' : binding_name , <EOL> '<STR_LIT>' : xmlelement , <EOL> } <EOL> self . binding = None <EOL> self . binding_options = None <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . name , self . binding , <EOL> self . binding_options ) <EOL> def __unicode__ ( self ) : <EOL> return '<STR_LIT>' % self . name <EOL> def send ( self , transport , operation , args , kwargs ) : <EOL> return self . binding . send ( <EOL> transport , self . binding_options , operation , args , kwargs ) <EOL> @ classmethod <EOL> def parse ( cls , wsdl , xmlelement ) : <EOL> """<STR_LIT>""" <EOL> name = qname_attr ( xmlelement , '<STR_LIT:name>' , wsdl . target_namespace ) <EOL> binding_name = qname_attr ( xmlelement , '<STR_LIT>' , wsdl . target_namespace ) <EOL> return cls ( name . text , binding_name = binding_name , xmlelement = xmlelement ) <EOL> def resolve ( self , definitions ) : <EOL> if self . _resolve_context is None : <EOL> return <EOL> binding = definitions . bindings . get ( <EOL> self . _resolve_context [ '<STR_LIT>' ] . text ) <EOL> if not binding : <EOL> return False <EOL> self . binding = binding <EOL> self . binding_options = binding . process_service_port ( <EOL> self . _resolve_context [ '<STR_LIT>' ] ) <EOL> self . _resolve_context = None <EOL> return True <EOL> class Service ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . ports = OrderedDict ( ) <EOL> self . name = name <EOL> self . _is_resolved = False <EOL> def __unicode__ ( self ) : <EOL> return '<STR_LIT>' % self . name . text <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . name . text , self . ports ) <EOL> def resolve ( self , definitions ) : <EOL> if self . _is_resolved : <EOL> return <EOL> unresolved = [ ] <EOL> for name , port in self . ports . items ( ) : <EOL> is_resolved = port . resolve ( definitions ) <EOL> if not is_resolved : <EOL> unresolved . append ( name ) <EOL> for name in unresolved : <EOL> del self . ports [ name ] <EOL> self . _is_resolved = True <EOL> def add_port ( self , port ) : <EOL> self . ports [ port . name ] = port <EOL> @ classmethod <EOL> def parse ( cls , definitions , xmlelement ) : <EOL> """<STR_LIT>""" <EOL> name = qname_attr ( xmlelement , '<STR_LIT:name>' , definitions . target_namespace ) <EOL> obj = cls ( name ) <EOL> for port_node in xmlelement . findall ( '<STR_LIT>' , namespaces = NSMAP ) : <EOL> port = Port . parse ( definitions , port_node ) <EOL> if port : <EOL> obj . add_port ( port ) <EOL> return obj </s>
