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

5

一些背景信息:我在解决身份验证问题后,在这里提出了这个问题。 我更喜欢开一个新的问题,以避免在原始问题中添加与原始问题无关的评论,并为其提供适当的可见性。

我正在使用运行在同一内部网络中且没有互联网访问权限的SOAP客户端。

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包含对位于企业内部网络之外的外部资源的导入('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客户端?
另外,基于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()
2个回答

5
我建议您进行自定义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中使用传输方式的方法描述在这里,但下面是使用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()

4

非常完美,謝謝 Michael!我會將我的解決方案作為問題的範例寫在裡面。 - Pintun

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接