WCF и Python

Есть ли какой-нибудь пример кода клиента cpython (не IronPython), который может вызвать службу Windows Communication Foundation (WCF)?

    7 Solutions collect form web for “WCF и Python”

    WCF должен раскрывать функциональность через протокол связи. Я думаю, что наиболее часто используемый протокол, вероятно, SOAP через HTTP. Предположим, что это то, что вы используете.

    Взгляните на эту главу в Dive Into Python . Он покажет вам, как сделать SOAP-вызовы.

    Я не знаю единого способа вызова службы WCF в Python, независимо от протокола связи.

    Я использовал пены .

    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 

    Это должно заставить вас начать. Я могу подключиться к открытым службам из WCF и слоя RESTful. Для того, чтобы сделать то, что вам нужно, необходимо провести некоторое массирование данных, особенно если вам нужно привязываться к нескольким пространствам имен.

    Просто чтобы помочь кому-то получить доступ к службе WCF SOAP 1.2 с помощью WS-Addressing с использованием пены. Основная проблема заключается в том, чтобы вводить имя действия в каждое сообщение.

    Этот пример для порта python 3 и suds https://bitbucket.org/jurko/suds .

    Пример использует пользовательскую аутентификацию на основе заголовков HTTP, я оставляю ее как есть.

    TODO: Автоматически получать api_direct_url из WSDL (теперь он жестко закодирован).

     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) в систему 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) 

    Я не знаю прямых примеров, но если служба WCF включена в REST, вы можете получить доступ к ней через POX (обычный старый XML) с помощью методов REST и т. Д. (Если у службы есть какие-либо). Если вы контролируете службу, вы можете также вывести конечные точки через REST.

    если вам нужна двоичная сериализованная связь по tcp, тогда рассмотрите решение, подобное Thrift.

    Даже если нет конкретного примера вызова WCF из Python, вы должны иметь возможность полностью использовать SOAP-совместимый сервис с WCF. Затем все, что вам нужно сделать, это найти несколько примеров того, как вызывать обычную службу SOAP с Python.

    Самое простое – использовать BasicHttpBinding в WCF, а затем вы можете поддерживать свои собственные сеансы, передавая токен сеанса с каждым запросом и ответом.

    TL; DR: для wsHttpBinding (SOAP 1.2) используйте zeep


    В случае, если у кого-то возникают проблемы с использованием пены (или suds-jurko, если на то пошло) с WCF и wsHttpBinding (это SOAP 1.2):

    • suds в значительной степени мертв (не может даже установить его на python 3)
    • suds-jurko кажется вроде мертвым. У версии 0.6 есть очень раздражающая бесконечная ошибка рекурсии (по крайней мере, на WSDL, представленном нашей службой), которая зафиксирована в подсказке, но это не выпущено, и это было 1,5 года (на момент написания этой статьи в феврале 17) с момента последнего совершить.
      Он работает на python 3, но не поддерживает SOAP 1.2. Ответ Совникова – попытка заставить его работать с 1.2, но мне не удалось заставить его работать для меня.
    • zeep, кажется, является текущим способом пойти и работать из коробки (я не связан с zeep, он просто работает для меня, и я провел несколько часов, стуча головой о кирпичную стену, пытаясь заставить плут работать). Для работы zeep конфигурация хоста службы WCF должна включать <security mode = "None" /> в узле wsHttpBinding.
     
    Interesting Posts for Van-Lav
    Python - лучший язык программирования в мире.