一些背景信息:在解决身份验证问题后,我在这里提出了这个问题。我更喜欢打开一个新的问题,以避免与原始问题无关的评论污染前一个问题,并给予它适当的可见性。
我正在开发一个与服务器在同一 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)
Run Code Online (Sandbox Code Playgroud)
问题: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',))
Run Code Online (Sandbox Code Playgroud)
问题:是否可以(并且有意义)在不访问外部资源的情况下创建 Zeep 客户端?
作为一个额外的细节,另一个用 Java 编写的基于 XML rpc ServiceFactory 的客户端似乎对此类问题更有弹性,即使没有可用的互联网连接,该服务也会创建(并工作)。真的需要从 xmlsoap.org 导入命名空间吗?
在 @mvt 回答后进行编辑:
因此,我选择了建议的解决方案,它允许我同时控制对外部资源的访问(阅读:禁止访问与托管端点的服务器不同的服务器)。
class MyTransport(zeep.Transport): …
Run Code Online (Sandbox Code Playgroud) 我试图在一个序列中选择使用Zeep,但我无法使这项工作成功.
这是我的代码:
#create trade ref element
tradeReference_element = client.get_element('ns4:tradeReference')
trade_ref_object = xsd.AnyObject(tradeReference_element, tradeReference_element(tradeId="NKY 12/08/17 C21000"))
#create export element
export_element = client.get_element('ns32:export')
export_element_object = xsd.AnyObject(export_element, export_element(_value_1={'tradeReference' : trade_ref_object}))
#use exportEntities service
client.service.exportEntities(export=export_element_object)
Run Code Online (Sandbox Code Playgroud)
我收到以下错误:
client.service.exportEntities(export=export_element_object)
Traceback (most recent call last):
File "<ipython-input-14-3cee211bc26a>", line 1, in <module>
client.service.exportEntities(export=export_element_object)
File "C:\ProgramData\Anaconda2\lib\site-packages\zeep\client.py", line 41, in __call__
self._op_name, args, kwargs)
File "C:\ProgramData\Anaconda2\lib\site-packages\zeep\wsdl\bindings\soap.py", line 110, in send
options=options)
File "C:\ProgramData\Anaconda2\lib\site-packages\zeep\wsdl\bindings\soap.py", line 68, in _create
serialized = operation_obj.create(*args, **kwargs)
File "C:\ProgramData\Anaconda2\lib\site-packages\zeep\wsdl\definitions.py", line 197, in create
return self.input.serialize(*args, **kwargs) …
Run Code Online (Sandbox Code Playgroud) 我正在尝试连接到 SOAP 网络服务并使用 Pandas 放在桌子上。
Zeep 给我这个列表:
[{
'ssPeca': '103',
'ssQtd': '1',
'ssUn': 'un'
}, {
'ssPeca': '291A',
'ssQtd': '8',
'ssUn': 'un'
}, {
'ssPeca': '406B',
'ssQtd': '8',
'ssUn': 'un'
}]
Run Code Online (Sandbox Code Playgroud)
我的代码是这样的:
client = zeep.Client(wsdl=wsdl)
pecas=client.service.TabelaPecas("C-160","CR")
pd.DataFrame.from_dict(pecas)
Run Code Online (Sandbox Code Playgroud)
并且该代码生成以下内容:
0 1 2
0 ssPeca ssQtd ssUn
1 ssPeca ssQtd ssUn
2 ssPeca ssQtd ssUn
Run Code Online (Sandbox Code Playgroud)
但我想要这个:
0 1 2
0 103 1 un
1 291A 8 un
2 406B 8 un
Run Code Online (Sandbox Code Playgroud)
有人可以帮忙吗?我只是python的初学者。
我正在构建一个 Python 应用程序,它使用 Python-zeep 访问第 3 方 SOAP API。我想使用模拟响应来实现一些单元测试,因为我并不总是有一个实时服务器来运行我的测试。
我是单元测试的新手,不太确定从哪里开始。我已经看到在请求库中使用模拟的示例,但不确定如何将其转换为 Zeep。
在我的一个模型上,我有一种方法可以从 SOAP API 获取所有 DevicePool。这是相关代码的摘录(我使用辅助方法来提供服务对象,因为我计划在许多其他方法中使用它)。
# Get Zeep Service Object to make AXL API calls
service = get_axl_client(self)
# Get list of all DevicePools present in the cluster
resp = service.listDevicePool(searchCriteria={'name': '%'},
returnedTags={'name': '', 'uuid': ''})
Run Code Online (Sandbox Code Playgroud)
我想模拟 resp 对象,但是这是 zeep.objects.ListDevicePoolRes 类型(一种基于 WSDL 解析的动态类型),我不能只用静态值实例化一个对象。
也许我在这里走错了路,必须更深入地模拟 zeep 库的一些内部结构并在 zeep 解析 XML 之前替换请求响应?
如果有人有类似的例子,将不胜感激。
我正在使用 python 3.6 和 zeep 3.4.0
Zeep 返回原始数据,我无法将其转换为 xml/json/pandas 对象。
我尝试使用 bs4 从 text1 获取表,但没有成功。序列化text1来获取json,也不走运。
from zeep import Client, Settings
settings = Settings(xml_huge_tree=True)
client = Client('http://www.cbr.ru/secinfo/secinfo.asmx?WSDL', settings=settings)
s = '2019-06-21T00:00:00'
with client.settings(raw_response=True):
result = (client.service.IDRepoRUBXML(s))
#print(dir(result))
text1 = (result.text)
print(text1)
#
#data = literal_eval(text1.decode('utf8'),)
def escape(t):
"""HTML-escape the text in `t`."""
return (t.replace("&","&").replace("<","<" ).replace( ">",">").replace("'","'").replace(""",'"'))
m = escape(text1)
print(m)
Run Code Online (Sandbox Code Playgroud)
我需要从 zeep 检索可读的 xml 或 json/pandas 表。
我有一个 zeep 显示为没有操作的 WSDL,但是当我在浏览器上拉起它时,WSDL 中有操作。作为参考,WSDL 需要证书。WSDL:
<?xml version="1.0" encoding="UTF-8"?>
<definitions targetNamespace="http://www.ercot.com/wsdl/2007-06/nodal/ewsConcrete" xmlns:ns2="http://schemas.xmlsoap.org/wsdl/http/" xmlns="http://schemas.xmlsoap.org/wsdl/" xmlns:ns1="http://schemas.xmlsoap.org/wsdl/soap12/" xmlns:ns4="http://www.w3.org/2006/05/addressing/wsdl" xmlns:ns3="http://schemas.xmlsoap.org/wsdl/jms/" xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/" xmlns:tns="http://www.ercot.com/wsdl/2007-06/nodal/ewsConcrete" xmlns:ns0="http://www.ercot.com/schema/2007-06/nodal/ews/message" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/" xmlns:wsu="http://www.docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd" xmlns:wsse="http://www.docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd">
<wsp:UsingPolicy xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" wsdl:Required="true" xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"/>
<wsp:Policy xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy"><sp:SecurityHeader MustManifestEncryption="true" MustPrepend="true" xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"/></wsp:Policy>
<wsp:Policy wsu:Id="SecurityTokens" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:ExactlyOne><wsp:All><sp:SecurityToken xmlns:_ns1="http://www.actional.com" xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"><sp:TokenType>_ns1:SSLClientCertificate</sp:TokenType></sp:SecurityToken></wsp:All><wsp:All><sp:SecurityToken xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"><sp:TokenType>sp:X509v3</sp:TokenType></sp:SecurityToken></wsp:All><wsp:All><sp:SecurityToken xmlns:_ns1="http://www.actional.com" xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"><sp:TokenType>_ns1:SSLClientCertificate</sp:TokenType></sp:SecurityToken><sp:SecurityToken xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"><sp:TokenType>sp:X509v3</sp:TokenType></sp:SecurityToken></wsp:All></wsp:ExactlyOne></wsp:Policy>
<wsp:Policy wsu:Id="SignedBody" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><sp:Integrity xmlns:sp="http://schemas.xmlsoap.org/ws/2002/12/secext"><sp:TokenInfo><sp:SecurityToken><sp:TokenType>sp:X509v3</sp:TokenType></sp:SecurityToken></sp:TokenInfo><sp:MessageParts>wsp:GetBody(.)</sp:MessageParts></sp:Integrity></wsp:Policy>
<wsp:Policy wsu:Id="Alerts_input_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SecurityTokens"/><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy>
<wsp:Policy wsu:Id="Alerts_output_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy>
<wsp:Policy wsu:Id="MarketInfo_input_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SecurityTokens"/><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy>
<wsp:Policy wsu:Id="MarketInfo_output_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy>
<wsp:Policy wsu:Id="MarketTransactions_input_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SecurityTokens"/><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy>
<wsp:Policy wsu:Id="MarketTransactions_output_policy" xmlns:wsp="http://schemas.xmlsoap.org/ws/2004/09/policy" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsp:PolicyReference URI="#SignedBody"/></wsp:Policy> …
Run Code Online (Sandbox Code Playgroud) 我有使用 zeep 创建肥皂客户端的代码。我的服务器不返回 wsdl 文件,但我在本地有它。
同步版本的工作原理如下:
import uuid
from os import path
import structlog
import zeep
logger = structlog.get_logger(__name__)
class SyncClient(object):
def __init__(self, ip_address: str):
self.ip_address = ip_address
self.port = 8080
self.soap_client = None
self.corrupt_timeseries_files = []
self.id = uuid.uuid4()
def connect_soap_client(self):
this_files_dir = path.dirname(path.realpath(__file__))
wsdl = 'file://{}'.format(path.join(this_files_dir, 'SOAPInterface.wsdl'))
transport = zeep.Transport(timeout=5, operation_timeout=3)
client = zeep.Client(wsdl, transport=transport)
location = "http://{}:{}".format(self.ip_address, str(self.port))
self.soap_client = client.create_service("{urn:webservices}SOAPInterface", location)
Run Code Online (Sandbox Code Playgroud)
然后 asyc 客户端看起来像这样:
class AsyncClient(object):
def __init__(self, ip_address: str):
self.ip_address = ip_address
self.port = 8080 …
Run Code Online (Sandbox Code Playgroud) 我从这样的 SOAP API 得到结果:
client = zeep.Client(wsdl=self.wsdl, transport=transport)
auth_header = lb.E("authenticate", self.login())
res = client.service.GetHouseProfile(region_id, page_number, reporting_period_id, _soapheaders=[auth_header])
Run Code Online (Sandbox Code Playgroud)
现在我需要解析 res 并得到结果。
>>> dir(res)
['__class__', '__contains__', '__deepcopy__', '__delattr__', '__delitem__', '__dict__', '__doc__', '__eq__', '__format__', '__getattribute__', '__getitem__', '__hash__', '__init__', '__iter__', '__len__', '__module__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__setitem__', '__sizeof__', '__str__', '__subclasshook__', '__values__', '__weakref__', '_xsd_type']
>>> type(res)
<class 'zeep.objects.GetHouseProfileSFResponse'>
>>> print(res.__str__()[0:100])
{
'data': {
'item': [
{
'house_id': 6465882L,
Run Code Online (Sandbox Code Playgroud)
如何从 res 中获取某个元素?
于是我找到了方法。看起来不是一个标准的决定,但它有效:
>>> res.__values__.get("data").__values__.get("item")[6].__values__.keys()
[u'house_id', u'house_profile_data', u'full_address', u'stage', u'state', u'emergency_date', u'emergency_number', u'emergency_reason', u'emergency_after', …
Run Code Online (Sandbox Code Playgroud) 我的 SOAP 服务将使用各种无效负载进行测试,以确认返回了适当的响应。
在进行测试之前,我想对我的服务进行自己的测试。一些测试涉及删除必填字段。我想使用 zeep 模拟这些测试,但 zeep 不允许我发送请求,zeep.exceptions.ValidationError
如果我想要发送的数据中不存在任何必填字段,则会给出一个请求。
我是否可以配置一些设置,以便 zeep 不会因缺少字段而引发错误并发送无效请求?
示例代码:
from zeep import Client
from datetime import datetime
wsdl_url = 'http://myservice.com/egservice?wsdl'
payload = {
'ServiceType': 'EgService',
'AvailabilityWindow': [
{'StartDateTime': datetime.now(),
'EndDateTime': datetime.now(),
'Validation': 'VALID'}],
'Confirmation': 'Confirmed',
'DateTimeStamp': datetime.now()
} # N.B No ContractID included
soap_client = Client(wsdl_url)
operation = 'myExampleOperation'
with soap_client.settings(raw_response=True):
response = soap_client.service[operation](**payload)
Run Code Online (Sandbox Code Playgroud)
wsdl 的相关部分:
<xs:complexType name="EgMessage">
<xs:sequence>
<xs:element name="ServiceType" type="tns:EgMessage_ServiceTypeType"/>
<xs:element name="ContractID" type="tns:EgMessage_ContractIDType"/>
<xs:element name="AUI" type="tns:EgMessage_AUIType" minOccurs="0"/>
<xs:element name="AvailabilityWindow" type="tns:AvailabilityWindowType" maxOccurs="unbounded"/>
<xs:element name="Confirmation" …
Run Code Online (Sandbox Code Playgroud) 我正在使用 zeep。我使用的服务不时产生超时错误,我想使用自动重试功能。
我正在尝试使用请求重试会话,但没有捕获超时异常,也没有尝试重试。
我设置了一个请求重试会话(如下),并设置了我的类的客户端属性(当前使用超时值导致错误):
session = requests_retry_session()
transport = Transport(session=session,timeout=20,operation_timeout=.001)
self.client = Client(self.wsdl_url,transport=transport)
Run Code Online (Sandbox Code Playgroud)
...
def requests_retry_session(
retries=10,
backoff_factor=0.3,
status_forcelist=(500, 502, 503, 504),
session=None,
) -> requests.Session:
session = session or requests.Session()
retry = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
Run Code Online (Sandbox Code Playgroud)
我收到 requests.exceptions.ReadTimeout (我假设这个错误与我在现实生活中遇到的错误相同,但是超时 20 秒,我的日志记录还不够好,尚无法确定)
会话未处理异常。
为什么不?
编辑:zeep 在某处发送 POST 请求,我在自己的调试中没有找到,但在 Requests 中激活调试日志显示了它。
默认情况下,重试不适用于 POST。
https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html
我设置了 method_whitelist=False 来改变它,现在我正在触发重试。