我想在 Python 中使用基于 WSDL SOAP 的 Web 服务。我查看了 Dive Into Python 代码,但 SOAPpy 模块在 Python 2.5 下不起作用。
我曾尝试使用 suds 部分工作,但与某些类型中断(suds.TypeNotFound:找不到类型:'item')。
我还查看了 Client,但这似乎不支持 WSDL。
我看过 ZSI,但它看起来非常复杂。有没有人有任何示例代码?
WSDL 是 https://ws.pingdom.com/soap/PingdomAPI.wsdl 并且可以与 PHP 5 SOAP 客户端一起正常工作。
我建议您看看 SUDS
“Suds 是一个用于使用 Web 服务的轻量级 SOAP python 客户端。”
有一个相对较新的库,它非常有前途,尽管文档仍然很差,但看起来非常干净和 pythonic:python zeep。
有关示例,另请参见 this answer。
我最近偶然发现了同样的问题。这是我的解决方案的概要:
需要的基本组成代码块
以下是您的客户端应用程序所需的基本代码块
会话请求部分:请求与提供者的会话会话身份验证部分:向提供者提供凭据客户端部分:创建客户端安全标头部分:将 WS-安全标头添加到客户端消耗部分:根据需要使用可用的操作(或方法)
你需要什么模块?
许多人建议使用 Python 模块,例如 urllib2 ;然而,没有一个模块可以工作——至少对于这个特定的项目是这样。
因此,这是您需要获取的模块列表。首先,您需要从以下链接下载并安装最新版本的 suds:
pypi.python.org/pypi/suds-jurko/0.4.1.jurko.2
此外,您需要分别从以下链接下载和安装 requests 和 suds_requests 模块(免责声明:我是新来这里发布的,所以我现在不能发布多个链接)。
pypi.python.org/pypi/requests pypi.python.org/pypi/suds_requests/0.1
成功下载并安装这些模块后,您就可以开始使用了。
编码
按照前面概述的步骤,代码如下所示: 导入:
import logging
from suds.client import Client
from suds.wsse import *
from datetime import timedelta,date,datetime,tzinfo
import requests
from requests.auth import HTTPBasicAuth
import suds_requests
会话请求和身份验证:
username=input('Username:')
password=input('password:')
session = requests.session()
session.auth=(username, password)
创建客户端:
client = Client(WSDL_URL, faults=False, cachingpolicy=1, location=WSDL_URL, transport=suds_requests.RequestsTransport(session))
添加 WS-Security 标头:
...
addSecurityHeader(client,username,password)
....
def addSecurityHeader(client,username,password):
security=Security()
userNameToken=UsernameToken(username,password)
timeStampToken=Timestamp(validity=600)
security.tokens.append(userNameToken)
security.tokens.append(timeStampToken)
client.set_options(wsse=security)
请注意,此方法创建图 1 中描述的安全标头。因此,您的实现可能会有所不同,具体取决于您正在使用的服务的所有者提供的正确安全标头格式。
消费相关方法(或操作):
result=client.service.methodName(Inputs)
记录:
此类实现中的最佳实践之一是记录以查看通信是如何执行的。万一出现问题,它使调试变得容易。以下代码执行基本日志记录。但是,除了代码中描述的内容之外,您还可以记录通信的许多方面。
logging.basicConfig(level=logging.INFO)
logging.getLogger('suds.client').setLevel(logging.DEBUG)
logging.getLogger('suds.transport').setLevel(logging.DEBUG)
结果:
这是我的情况的结果。请注意,服务器返回 HTTP 200。这是 HTTP 请求响应的标准成功代码。
(200, (collectionNodeLmp){
timestamp = 2014-12-03 00:00:00-05:00
nodeLmp[] =
(nodeLmp){
pnodeId = 35010357
name = "YADKIN"
mccValue = -0.19
mlcValue = -0.13
price = 36.46
type = "500 KV"
timestamp = 2014-12-03 01:00:00-05:00
errorCodeId = 0
},
(nodeLmp){
pnodeId = 33138769
name = "ZION 1"
mccValue = -0.18
mlcValue = -1.86
price = 34.75
type = "Aggregate"
timestamp = 2014-12-03 01:00:00-05:00
errorCodeId = 0
},
})
suds_request
在安装时会失败,因此如果您使用 suds-jurko
fork,您可以安装适用于 jurko 的 suds 版本的 suds_request
:pip install git+https://github.com/chrcoe/suds_requests.git@feature/python3_suds_jurko
Zeep 是一个不错的 Python SOAP 库,符合您的要求:http://docs.python-zeep.org
现在(截至 2008 年),所有可用于 Python 的 SOAP 库都很糟糕。如果可能,我建议避免使用 SOAP。上次我们被迫使用 Python 中的 SOAP Web 服务时,我们用 C# 编写了一个包装器,它一方面处理 SOAP,另一方面发出 COM。
我会定期为此寻找令人满意的答案,但到目前为止还没有运气。我使用soapUI +请求+体力劳动。
上次我需要这样做时我放弃并使用了 Java,并且在我最后一次想要这样做时简单地放弃了几次,但这并不是必需的。
去年通过 Project Place 的 RESTful API 成功使用 requests 库后,我突然想到,也许我可以手动滚动我想以类似方式发送的 SOAP 请求。
事实证明这并不太难,但它既费时又容易出错,特别是如果字段名称不一致(我今天正在处理的那个有'jobId',JobId'和'JobID'。我使用soapUI加载WSDL,以便更容易提取端点等并执行一些手动测试。到目前为止,我很幸运没有受到我正在使用的任何 WSDL 更改的影响。
这不是真的 SOAPpy 不适用于 Python 2.5 - 它可以工作,尽管它非常简单并且非常非常基本。如果您想与任何更复杂的 Web 服务交谈,ZSI 是您唯一的朋友。
我在 http://www.ebi.ac.uk/Tools/webservices/tutorials/python 找到了真正有用的演示 - 这确实帮助我了解了 ZSI 的工作原理。
SOAPpy 现在已经过时了,AFAIK 被 ZSL 取代。这是一个有争议的问题,因为我无法在 Python 2.5 或 Python 2.6 上让任何一个工作,更不用说编译了
#!/usr/bin/python
# -*- coding: utf-8 -*-
# consume_wsdl_soap_ws_pss.py
import logging.config
from pysimplesoap.client import SoapClient
logging.config.dictConfig({
'version': 1,
'formatters': {
'verbose': {
'format': '%(name)s: %(message)s'
}
},
'handlers': {
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'verbose',
},
},
'loggers': {
'pysimplesoap.helpers': {
'level': 'DEBUG',
'propagate': True,
'handlers': ['console'],
},
}
})
WSDL_URL = 'http://www.webservicex.net/stockquote.asmx?WSDL'
client = SoapClient(wsdl=WSDL_URL, ns="web", trace=True)
client['AuthHeaderElement'] = {'username': 'someone', 'password': 'nottelling'}
#Discover operations
list_of_services = [service for service in client.services]
print(list_of_services)
#Discover params
method = client.services['StockQuote']
response = client.GetQuote(symbol='GOOG')
print('GetQuote: {}'.format(response['GetQuoteResult']))
不定期副业成功案例分享