A-A+

python 最好用的SOAP客户端 Zeep

2019年01月25日 18:14 汪洋大海暂无评论共6321字 (阅读5,441 views次)

无论你是python几，都建议你用zeep来链接你的soap接口，因为它是最新的而且还有人维护的库。

zeep客户端对象

它Client是与SOAP服务器交互的主要接口。它提供了一个service引用客户端默认绑定的属性（通过ServiceProxy对象）。通过传递service_name和启动客户端时，可以指定默认绑定port_name。否则，该服务中的第一个服务和第一个端口将用作默认值。

缓存WSDL和XSD文件

初始化客户端时，它将自动检索作为参数传递的WSDL文件。此WSDL文件通常引用各种其他WSDL和XSD文件。默认情况下，Zeep不会缓存这些文件，但建议出于性能原因启用此文件。

请参阅缓存如何启用此功能。为了使其易于使用，zeep.CachingClient()自动创建启用了SqliteCache的Transport对象。

配置客户端

Client类接受配置客户端的设置参数。您可以使用以下代码初始化对象：

from zeep import Client, Settings

settings = Settings(strict=False, xml_huge_tree=True)
client = Client('http://my-wsdl/wsdl', settings=settings)

始终可以通过客户端访问设置对象 client.settings。例如：

with client.settings(raw_response=True):
    response = client.service.myoperation()

有关更多信息，请参阅设置。

严格模式

默认情况下，zeep将以“严格”模式运行。如果您使用严格设置使用不符合标准的SOAP服务器，则可以禁用此功能。请参阅设置。禁用严格模式将更改以下行为：

在启用恢复模式的情况下解析XML

xsd：sequences中允许缺少非可选元素

请注意，禁用严格模式应该被视为最后的手段，因为它可能导致XML和返回的响应之间的数据丢失。

ServiceProxy对象

ServiceProxy对象是一个简单的对象，它将检查所请求的属性或项目是否存在操作。如果操作存在，那么它将返回一个OperationProxy对象（可调用），该对象负责调用绑定上的操作。

from zeep import Client
from zeep import xsd

client = Client('http://my-endpoint.com/production.svc?wsdl')

# service is a ServiceProxy object.  It will check if there
# is an operation with the name `X` defined in the binding
# and if that is the case it will return an OperationProxy
client.service.X()

# The operation can also be called via an __getitem__ call.
# This is useful if the operation name is not a valid
# python attribute name.
client.service['X-Y']()

使用非默认绑定

正如默认提到的那样，Zeep选择WSDL中的第一个绑定作为默认绑定。这种绑定可以通过client.service。要使用特定绑定，您可以bind()在客户端对象上使用该方法：

from zeep import Client
from zeep import xsd

client = Client('http://my-endpoint.com/production.svc?wsdl')

service2 = client.bind('SecondService', 'Port12')
service2.someOperation(myArg=1)

创建新的ServiceProxy对象

在某些情况下，您需要更改WSDL中定义的SOAP地址，或者WSDL不定义任何服务元素。这可以通过使用该Client.create_service()方法创建新的ServiceProxy来完成。

from zeep import Client
from zeep import xsd

client = Client('http://my-endpoint.com/production.svc?wsdl')
service = client.create_service(
    '{http://my-target-namespace-here}myBinding',
    'http://my-endpoint.com/acceptance/')

service.submit('something')

创建原始XML文档

如果希望zeep构建并返回XML而不是将其发送到服务器，则可以使用该Client.create_message()调用。它需要ServiceProxy作为第一个参数，操作名称作为第二个参数。

from zeep import Client

client = Client('http://my-endpoint.com/production.svc?wsdl')
node = client.create_message(client.service, 'myOperation', user='hi')

运输

如果需要更改缓存，超时或TLS（或SSL）验证等选项，则需要自己创建Transport类的实例。

注意

安全套接字层（SSL）已被弃用，转而使用传输层安全性（TLS）。2011年禁止使用SSL 2.0，2015年6月禁止使用SSL 3.0。

TLS验证

如果您需要验证TLS连接（如果您的主机有自签名证书），最好的方法是创建一个requests.Session实例并将信息添加到该Session，这样它就会保持持久性：

from requests import Session
from zeep import Client
from zeep.transports import Transport

session = Session()
session.verify = 'path/to/my/certificate.pem'
transport = Transport(session=session)
client = Client(
    'http://my.own.sslhost.local/service?WSDL',
    transport=transport)

暗示

确保您引用的证书是CA_BUNDLE，这意味着它包含根CA和中间CA. 接受的只是X.509 ASCII文件（.pem有时是文件扩展名.crt）。如果您有两个不同的文件，则必须手动将它们合并为一个文件。

或者，如果您只想使用TLS客户端证书，则session.verify可以使用而不是使用session.cert。

要禁用TLS验证（不推荐！），您需要设置 verify为False。

session = Session()
session.verify = False

甚至更简单的方式：

client.transport.session.verify = False

请记住：这应该仅用于测试目的。Python urllib3 将用一个警告你InsecureRequestWarning。

有关requests.Session详细信息，请参阅

会话超时

要设置传输超时，请使用timeout选项。默认超时为300秒：

from zeep import Client
from zeep.transports import Transport

transport = Transport(timeout=10)
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    transport=transport)

使用HTTP或SOCKS代理

默认情况下，zeep requests用作传输层，允许使用以下proxies属性定义代理requests.Session：

from zeep.import Client

client = Client(
    'http://my.own.sslhost.local/service?WSDL')

client.transport.session.proxies = {
    # Utilize for all http/https connections
    'http': 'foo.bar:3128',
    'https': 'foo.bar:3128',
    # Utilize for certain URL
    'http://specific.host.example': 'foo.bar:8080',
    # Or use socks5 proxy (requires requests[socks])
    'https://socks5-required.example': 'socks5://foo.bar:8888',
}

为了使用SOCKS代理，需要使用其他包安装请求（例如）。pip install -Urequests[socks]

缓存

默认情况下，zeep不使用缓存后端。为了性能优势，建议使用SqliteCache后端。默认情况下，它将WSDL和XSD文件缓存1小时。要使用缓存后端，请使用以下命令初始化客户端：

from zeep import Client
from zeep.cache import SqliteCache
from zeep.transports import Transport

transport = Transport(cache=SqliteCache())
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    transport=transport)

更改SqliteCache设置可以通过以下方式完成：

from zeep import Client
from zeep.cache import SqliteCache
from zeep.transports import Transport
cache = SqliteCache(path='/tmp/sqlite.db', timeout=60)
transport = Transport(cache=cache)
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    transport=transport)

另一种选择是使用InMemoryCache后端。它在内部使用全局字典来存储具有相应内容的URL。

HTTP认证

虽然某些提供程序在SOAP消息的标头中包含安全功能，但其他提供程序使用HTTP身份验证标头。在后一种情况下，您可以requests.Session使用auth set 创建一个对象并将其传递给Transport类。

from requests import Session
from requests.auth import HTTPBasicAuth  # or HTTPDigestAuth, or OAuth1, etc.
from zeep import Client
from zeep.transports import Transport

session = Session()
session.auth = HTTPBasicAuth(user, password)
client = Client('http://my-endpoint.com/production.svc?wsdl',
    transport=Transport(session=session))

调试

要查看发送到远程服务器的SOAP XML消息以及收到的响应，可以将zeep.transports模块的Python记录器级别设置为DEBUG 。从0.15开始，这也可以通过HistoryPlugin实现。

import logging.config

logging.config.dictConfig({
    'version': 1,
    'formatters': {
        'verbose': {
            'format': '%(name)s: %(message)s'
        }
    },
    'handlers': {
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'verbose',
        },
    },
    'loggers': {
        'zeep.transports': {
            'level': 'DEBUG',
            'propagate': True,
            'handlers': ['console'],
        },
    }
})

-------下面放几个案例。

from requests.auth import HTTPBasicAuth  # or HTTPDigestAuth, or OAuth1, etc.
from requests import Session
from zeep import Client
from zeep.transports import Transport
 
session = Session()
session.auth = HTTPBasicAuth(user, password)
client = Client('http://my-endpoint.com/production.svc?wsdl',
            transport=Transport(session=session))

------------------------------

import zeep
from zeep.client import Client
from zeep.transports import Transport
from zeep.cache import SqliteCache
from zeep.wsse.username import UsernameToken
import requests
from types import *
from datetime import datetime
from datetime import timedelta
 
 
USERNAME = '94a8e8abffb20dfc4826221c2b8ba1d058a46c3cd57e11487170620'
PASSWORD = '0f503938100c2839c2f2fafe0cc6e406'
wsdl = 'https://webservices.chargepoint.com/cp_api_5.0.wsdl'
 
transport = Transport(cache=SqliteCache())
client = zeep.Client(wsdl=wsdl, wsse=UsernameToken(USERNAME, PASSWORD))
#response = client.service.getStations(searchQuery = {'stationName':'AHM TORRANCE / 27-28'})
#print response
today = datetime(2017,06,23,19,00,00)
 
## client.service.getStations(searchQuery = {})
print client.service.getLoad(searchQuery =  {'stationID':'1:115881'})
print client.service.getStations(searchQuery = {'stationID':'1:115791'})

蜗居

窄小蜗居，虽非富贵王侯宅；清闲螺径，也异寻常百姓家。 woj → 蜗居

python 最好用的SOAP客户端 Zeep

zeep客户端对象

缓存WSDL和XSD文件

配置客户端

严格模式

ServiceProxy对象

使用非默认绑定

创建新的ServiceProxy对象

创建原始XML文档

运输

TLS验证

会话超时

使用HTTP或SOCKS代理

缓存

HTTP认证

调试

给我留言取消回复

zeep客户端对象

缓存WSDL和XSD文件

配置客户端

严格模式

ServiceProxy对象

使用非默认绑定

创建新的ServiceProxy对象

创建原始XML文档

运输

TLS验证

会话超时

使用HTTP或SOCKS代理

缓存

HTTP认证

调试

布施恩德可便相知重

微信扫一扫打赏

支付宝扫一扫打赏

给我留言取消回复