一个小上下文:在解决了一个身份验证问题之后,我打开了这个问题引发了这里。我更喜欢打开一个新的,以避免污染以前的评论与原来的问题,并给予它适当的可见度。
我正在运行一个SOAP客户端,运行在与服务器相同的intranet上,没有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包含一个导入到内部网外部的外部资源(‘导入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的用Java编写的客户机似乎更能适应这种问题,即使没有可用的internet连接,服务也会被创建(并且工作)。是否真的需要从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()发布于 2016-10-27 08:58:30
您可以创建tranport类的自己的子类,并向load()方法中添加额外的逻辑,以便从文件系统重定向/加载特定的url。
代码很简单,我认为:https://github.com/mvantellingen/python-zeep/blob/master/src/zeep/transports.py :-)
发布于 2018-09-14 16:10:53
我建议您对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()https://stackoverflow.com/questions/40220134
复制相似问题