Gibt es einen Beispielcode für eine cpython (nicht IronPython) Client, der den Windows Communication Foundation (WCF) Dienst aufrufen kann?
Antworten
Zu viele Anzeigen?Ich habe schäumt .
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
Damit sollten Sie anfangen können. Ich bin in der Lage, eine Verbindung zu exponierten Diensten von WCF und einer RESTful-Schicht herzustellen. Es müssen einige Daten massiert werden, um zu tun, was Sie brauchen, vor allem, wenn Sie an mehrere Namespaces binden müssen.
TL;DR: Für wsHttpBinding (SOAP 1.2) verwenden zeep
Für den Fall, dass jemand Probleme mit suds (oder suds-jurko für diese Angelegenheit) mit WCF und wsHttpBinding (die SOAP 1.2 ist) hat:
- suds ist so gut wie tot (kann nicht einmal mit Pip auf Python 3 installiert werden)
- suds-jurko scheint irgendwie tot zu sein. Die Version 0.6 hat einen sehr ärgerlichen Fehler mit unendlicher Rekursion (zumindest bei der WSDL, die von unserem Dienst bereitgestellt wird), der im Tipp behoben ist, aber das ist nicht veröffentlicht und es ist 1,5 Jahre her (zum Zeitpunkt dieses Schreibens im Februar 17) seit der letzten Übertragung.
Es funktioniert mit Python 3, unterstützt aber nicht SOAP 1.2. Sovetnikovs Antwort ist ein Versuch, es mit 1.2 zum Laufen zu bringen, aber ich habe es nicht geschafft, es für mich zum Laufen zu bringen. - zeep scheint der derzeitige Weg zu sein und funktioniert sofort (ich bin nicht mit zeep verbunden, es funktioniert einfach für mich und ich habe mehrere Stunden damit verbracht, meinen Kopf gegen eine Mauer zu schlagen, um suds zum Laufen zu bringen).
Damit zeep funktioniert, muss die Konfiguration des WCF-Service-Hosts <security mode="None"/> unter dem Knoten wsHttpBinding enthaltenTatsächlich scheint zeep zu unterstützen Benutzernamen und Signatur (x509) basierte WS-SE aber ich habe das noch nicht ausprobiert und kann daher nicht sagen, ob es damit Probleme gibt.
WCF muss Funktionen über ein Kommunikationsprotokoll bereitstellen. Ich denke, das am häufigsten verwendete Protokoll ist wahrscheinlich SOAP über HTTP. Nehmen wir an, das ist was Sie dann verwenden.
Werfen Sie einen Blick auf dieses Kapitel in Dive Into Python . Es zeigt Ihnen, wie Sie SOAP-Aufrufe tätigen.
Ich kenne keine einheitliche Methode zum Aufrufen eines WCF-Dienstes in Python, unabhängig von der Kommunikation Protokoll.
Ich möchte jemandem helfen, auf einen WCF SOAP 1.2 Dienst mit WS-Adressierung unter Verwendung von Suds zuzugreifen. Das Hauptproblem ist, den Namen der Aktion in jede Nachricht zu injizieren.
Dieses Beispiel für python 3 und suds port https://bitbucket.org/jurko/suds .
Das Beispiel verwendet eine benutzerdefinierte Authentifizierung auf der Grundlage von HTTP-Headern, ich lasse es so wie es ist.
TODO: Automatisch api_direct_url aus WSDL holen (zur Zeit ist es hart kodiert).
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)
- See previous answers
- Weitere Antworten anzeigen