WCF và Python

Question

Có mã ví dụ nào của ứng dụng khách cpython (không phải IronPython) có thể gọi dịch vụ Windows Communication Foundation (WCF) không?

Answer 1

WCF cần hiển thị chức năng thông qua giao thức truyền thông. Tôi nghĩ rằng giao thức được sử dụng phổ biến nhất có lẽ là SOAP qua HTTP. Giả sử đó là những gì bạn đang sử dụng sau đó.

Hãy xem this chapter in Dive Into Python. Nó sẽ cho bạn thấy làm thế nào để thực hiện cuộc gọi SOAP.

Tôi biết không có cách nào thống nhất gọi dịch vụ WCF bằng Python, bất kể giao thức là gì.

Answer 2

Thậm chí nếu không có ví dụ cụ thể về việc gọi WCF từ Python, bạn sẽ có thể thực hiện dịch vụ tuân thủ SOAP hoàn toàn với WCF. Sau đó, tất cả những gì bạn phải làm là tìm một số ví dụ về cách gọi một dịch vụ SOAP bình thường từ Python.

Điều đơn giản nhất là sử dụng BasicHttpBinding trong WCF và sau đó bạn có thể hỗ trợ phiên của riêng mình bằng cách chuyển mã thông báo phiên với mỗi yêu cầu và phản hồi.

Answer 3

Tôi không biết bất kỳ ví dụ trực tiếp nào, nhưng nếu dịch vụ WCF được kích hoạt REST, bạn có thể truy cập thông qua POX (Plain Old XML) qua phương thức REST/etc (nếu dịch vụ có). Nếu bạn có quyền kiểm soát dịch vụ, bạn cũng có thể phơi bày điểm cuối thông qua REST.

Answer 4

nếu bạn cần giao tiếp tuần tự hóa nhị phân trên tcp thì hãy xem xét triển khai giải pháp như Tiết kiệm.

Answer 5

Tôi đã sử dụng 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 

Điều đó sẽ giúp bạn bắt đầu. Tôi có thể kết nối với các dịch vụ tiếp xúc từ WCF và một lớp RESTful. Có cần phải có một số dữ liệu mát xa để giúp làm những gì bạn cần, đặc biệt là nếu bạn cần phải liên kết với một số không gian tên.

Answer 6

Chỉ để giúp một người nào đó truy cập dịch vụ WCF SOAP 1.2 với WS-Addressing bằng cách sử dụng suds. Vấn đề chính là đưa tên hành động vào mọi thư.

Ví dụ này cho cổng python 3 và suds https://bitbucket.org/jurko/suds.

Ví dụ sử dụng xác thực tùy chỉnh dựa trên tiêu đề HTTP, tôi để nguyên trạng thái này.

TODO: Tự động nhận api_direct_url từ WSDL (lúc này mã được mã hóa cứng).

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) 

Answer 7

TL; DR: Đối với wsHttpBinding (SOAP 1.2) sử dụng zeep

---

Trong trường hợp ai đó đang gặp khó khăn trong sử dụng bọt (hoặc bọt-jurko cho rằng vấn đề) với WCF và wsHttpBinding (đó là SOAP 1.2):

• bọt là khá nhiều người chết (thậm chí không thể pip cài đặt nó trên python 3)
• bọt-jurko dường như loại của người chết. Bản phát hành 0.6 có lỗi đệ quy vô hạn rất khó chịu (ít nhất là trên WSDL mà dịch vụ của chúng tôi tiếp xúc) được sửa trong đầu nhưng không được phát hành và nó đã được 1.5 năm (tại thời điểm viết bài này trong Feb'17) kể từ lần commit cuối cùng.
  Nó hoạt động trên trăn 3 nhưng không hỗ trợ SOAP 1.2. Sovetnikov's answer là một nỗ lực để làm cho nó hoạt động với 1,2 nhưng tôi đã không quản lý để làm cho nó làm việc cho tôi.
• zeep dường như là cách hiện tại để đi và làm việc ra khỏi hộp (Tôi không liên kết với zeep, nó chỉ hoạt động cho tôi và tôi đã dành vài giờ đập đầu của tôi chống lại một bức tường gạch cố gắng để làm cho suds làm việc) . Để zeep hoạt động, cấu hình máy chủ dịch vụ WCF phải bao gồm < chế độ bảo mật = "Không"/> trong nút wsHttpBinding.