WCF et Python

J'ai utilisé suds.

from suds.client import Client

print "Connecting to Service..."
wsdl = "http://serviceurl.com/service.svc?WSDL"
client = Client(wsdl)
result = client.service.Method(variable1, variable2)
print result

Qui devrait vous obtenir a commencé. Je suis en mesure de se connecter à des services exposés à partir de la WCF et une bonne couche. Il doit y avoir des données de massage pour aider à faire ce que vous avez besoin, surtout si vous avez besoin de se lier à plusieurs espaces de noms.

Juste pour aider une personne à accéder à la WCF SOAP 1.2 service avec WS-Addressing à l'aide de la mousse.
Principal problème consiste à injecter au nom de l'action dans chaque message.

Cet exemple pour python 3 et la mousse de port https://bitbucket.org/jurko/suds.

Exemple utilise personnalisé authentification basée sur les en-têtes HTTP, je laisse comme c'est.

TODO: obtenir Automatiquement api_direct_url à partir de WSDL (à maintenant, il est codé en dur).

from suds.plugin import MessagePlugin
from suds.sax.text import Text
from suds.wsse import Security, UsernameToken
from suds.sax.element import Element
from suds.sax.attribute import Attribute
from suds.xsd.sxbasic import Import
api_username = 'some'
api_password = 'none'
class api(object):
api_direct_url = 'some/mex'
api_url = 'some.svc?singleWsdl|Wsdl'
NS_WSA = ('wsa', 'http://www.w3.org/2005/08/addressing')
_client_instance = None
@property
def client(self):
if self._client_instance:
return self._client_instance
from suds.bindings import binding
binding.envns = ('SOAP-ENV', 'http://www.w3.org/2003/05/soap-envelope')
api_inst = self
class _WSAPlugin(MessagePlugin):
def marshalled(self, context):
api_inst._marshalled_message(context)
self._client_instance = Client(self.api_url,
plugins=[_WSAPlugin()],
headers={'Content-Type': 'application/soap+xml',
'login':api_username,
'password': api_password}
)
headers = []
headers.append(Element('To', ns=self.NS_WSA).setText(self.api_direct_url))
headers.append(Element('Action', ns=self.NS_WSA).setText('Blank'))
self._client_instance.set_options(soapheaders=headers)
cache = self._client_instance.options.cache
cache.setduration(days=10)
return self._client_instance
def _marshalled_message(self, context):
def _children(r):
if hasattr(r, 'children'):
for c in r.children:
yield from _children(c)
yield c
for el in _children(context.envelope):
if el.name == 'Action':
el.text = Text(self._current_action)
return
_current_action = None
def _invoke(self, method, *args):
try:
self._current_action = method.method.soap.action.strip('"')
return method(*args)
finally:
self._current_action = None
def GetRequestTypes(self):
return self._invoke(self.client.service.GetRequestTypes)[0]
def GetTemplateByRequestType(self, request_type_id):
js = self._invoke(self.client.service.GetTemplateByRequestType, request_type_id)
return json.loads(js)
def GetRequestStatus(self, request_guid):
return self._invoke(self.client.service.GetRequestStatus, request_guid)
def SendRequest(self, request_type_id, request_json):
r = json.dumps(request_json, ensure_ascii=False)
return self._invoke(self.client.service.SendRequest, request_type_id, r)

si vous avez besoin d'binaire sérialisé communication sur tcp puis envisager de mettre en œuvre des solutions comme l'Épargne.