标签: zeep

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

一些背景信息:在解决身份验证问题后,我在这里提出了这个问题。我更喜欢打开一个新的问题,以避免与原始问题无关的评论污染前一个问题,并给予它适当的可见性。

我正在开发一个与服务器在同一 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)

python soap wsdl xml-rpc zeep

5
推荐指数
2
解决办法
7650
查看次数

Python Zeep - 如何使用序列 - >选择

我试图在一个序列中选择使用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)

python xsd soap wsdl zeep

5
推荐指数
0
解决办法
870
查看次数

Python,Zeep 对熊猫的响应

我正在尝试连接到 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 pandas zeep

5
推荐指数
2
解决办法
3486
查看次数

有人有使用 Python Zeep 和 Mock 对 SOAP API 进行单元测试的示例吗?

我正在构建一个 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 unit-testing mocking zeep

5
推荐指数
1
解决办法
1894
查看次数

有没有一种简单的方法可以将 zeep 响应转换为 json、pandas、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("&amp;","&").replace("&lt;","<" ).replace( "&gt;",">").replace("&#39;","'").replace("&quot;",'"'))

m = escape(text1)
print(m)

Run Code Online (Sandbox Code Playgroud)

我需要从 zeep 检索可读的 xml 或 json/pandas 表。

python json wsdl pandas zeep

5
推荐指数
2
解决办法
8428
查看次数

如何处理 Zeep 无法识别的 WSDL 操作

我有一个 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)

python xml soap wsdl zeep

5
推荐指数
0
解决办法
717
查看次数

如何使用 wsdl 文件创建异步 zeep 客户端?

我有使用 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)

python python-3.x python-asyncio zeep

5
推荐指数
1
解决办法
2045
查看次数

python soap zeep模块获取结果

我从这样的 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)

python soap zeep

4
推荐指数
2
解决办法
1万
查看次数

Zeep 发送缺少强制标签的请求

我的 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)

python xsd soap wsdl zeep

4
推荐指数
1
解决办法
1790
查看次数

如何使用 python zeep 进行重试处理?我正在使用请求重试会话,但未处理异常

我正在使用 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 来改变它,现在我正在触发重试。

python python-requests zeep

4
推荐指数
1
解决办法
1682
查看次数