A-A+

python 最好用的SOAP客户端 Zeep

2019年01月25日 18:14 汪洋大海 暂无评论 阅读 555 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验证(不推荐!),您需要设置 verifyFalse

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'],
        },
    }
})

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

1
2
3
4
5
6
7
8
9
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))

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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'})

布施恩德可便相知重

微信扫一扫打赏

支付宝扫一扫打赏

×
标签:

给我留言