带有Zeep的Python SOAP客户端 - 导入命名空间

Posted 2021-04-14

一个小小的背景：在解决身份验证问题后，我正在打开here这个问题。我更喜欢打开一个新的，以避免污染前一个与原始问题无关的评论，并给它适当的可见性。

我正在使用与服务器在同一个Intranet中运行的SOAP客户端，无需访问Internet。

from requests.auth import HTTPBasicAuth
from zeep import Client
from zeep.transports import Transport

wsdl = 'http://mysite.dom/services/MyWebServices?WSDL'
client = Client(wsdl, transport=HTTPBasicAuth('user','pass'), cache=None)

问题：WSDL包含对位于Intranet外部的外部资源的导入（'import namespace =“schemas.xmlsoap.org/soap/encoding/”'），因此Zeep Client实例化失败：

Exception: HTTPConnectionPool(host='schemas.xmlsoap.org', port=80): Max retries exceeded with url: /soap/encoding/ (Caused by NewConnectionError('<requests.packages.urllib3.connection.HTTPConnection object at 0x7f3dab9d30b8>: Failed to establish a new connection: [Errno 110] Connection timed out',))

问题：在不访问外部资源的情况下创建Zeep客户端是否可能（并且有意义）？

作为一个额外的细节，基于XML rpc ServiceFactory的另一个用Java编写的客户端似乎对这类问题更具弹性，即使没有可用的互联网连接，服务也会被创建（并且有效）。是否真的需要从xmlsoap.org导入命名空间？

从@mvt回答后编辑：

所以，我选择了提议的解决方案，它允许我同时控制对外部资源的访问（读取：禁止访问与托管端点的服务器不同的服务器）。

class MyTransport(zeep.Transport):
    def load(self, url):
        if not url:
            raise ValueError("No url given to load")
        parsed_url = urlparse(url)
        if parsed_url.scheme in ('http', 'https'):
            if parsed_url.netloc == "myserver.ext":
                response = self.session.get(url, timeout=self.load_timeout)
                response.raise_for_status()
                return response.content
            elif url == "http://schemas.xmlsoap.org/soap/encoding/":
                url = "/some/path/myfile.xsd"
            else:
                raise
        elif parsed_url.scheme == 'file':
            if url.startswith('file://'):
                url = url[7:]
        with open(os.path.expanduser(url), 'rb') as fh:
            return fh.read()

答案

您可以创建自己的传输类子类，并向load（）方法添加其他逻辑，以便从文件系统重定向/加载特定的URL。

我认为代码很简单：https://github.com/mvantellingen/python-zeep/blob/master/src/zeep/transports.py :-)

另一答案

我建议您自定义覆盖URL并从超类调用load（）。这样，如果超类代码更改（它有），您将不需要重构您的CustomTransport类。

from zeep.transports import Transport

class CustomTransport(Transport):
    def load(self, url):
        # Custom URL overriding to local file storage
        if url and url == "http://schemas.xmlsoap.org/soap/encoding/":
            url = "/path/to/schemas.xmlsoap.org.xsd"

        # Call zeep.transports.Transport's load()
        return super(CustomTransport, self).load(url)

在here中描述了在zeep中使用Transports的方法，但这里有一个使用CustomTransport的简单示例：

from requests import Session
from requests.auth import HTTPBasicAuth
from zeep import Client

session = Session()
client = Client('http://example.com/production.svc?wsdl', transport=CustomTransport(session=session))
client.service.foo()