使用 Zeep 的 Python SOAP 客户端 - 导入命名空间

2024-03-14

一些背景:我在打开这个问题时出现了here https://stackoverflow.com/questions/40173217/python-soap-client-with-zeep-authentication,解决身份验证问题后。我更喜欢打开一个新的问题,以避免与原始问题无关的评论污染前一个问题,并给予它适当的可见性。

我正在开发一个与服务器在同一 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 客户端实例化失败并显示:

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 客户端?

作为一个额外的细节,另一个用 Java 编写的基于 XML rpc ServiceFactory 的客户端似乎对此类问题更有弹性,即使没有可用的互联网连接,该服务也会创建(并工作)。 真的需要从 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()

我建议您自定义重写 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)

描述了zeep中Transports的使用方法here https://python-zeep.readthedocs.io/en/master/transport.html,但这里是使用 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()
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

使用 Zeep 的 Python SOAP 客户端 - 导入命名空间 的相关文章

随机推荐