如何在Python的Zeep中传递SOAP请求的Any类型参数

问题背景 #

我有一个WSDL文件，所有SOAP操作的核心元素（Element）都使用了"any"类型（已精简WSDL内容）。想用Python 3的zeep库生成指定格式的SOAP请求Payload，但始终无法正确处理这个Any类型。

WSDL相关片段 #

<?xml version="1.0" encoding="utf-8"?>
<definitions targetNamespace="urn:xtk:queryDef" xmlns="http://schemas.xmlsoap.org/wsdl/" xmlns:s="http://www.w3.org/2001/XMLSchema" xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/" xmlns:tns="urn:xtk:queryDef" xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/">
  <types>
    <s:schema elementFormDefault="qualified" targetNamespace="urn:xtk:queryDef">
        <s:complexType name="Element">
            <s:sequence>
                <s:any processContents="lax"/>
            </s:sequence>
        </s:complexType>
      <s:element name="ExecuteQuery">
        <s:complexType>
          <s:sequence>
            <s:element maxOccurs="1" minOccurs="1" name="sessiontoken" type="s:string" />
            <s:element maxOccurs="1" minOccurs="1" name="entity" type="tns:Element" />
          </s:sequence>
        </s:complexType>
      </s:element>
      <s:element name="ExecuteQueryResponse">
        <s:complexType>
          <s:sequence>
            <s:element maxOccurs="1" minOccurs="1" name="pdomOutput" type="tns:Element" />
          </s:sequence>
        </s:complexType>
      </s:element>
    </s:schema>
  </types>
  <message name="ExecuteQueryIn">
    <part element="tns:ExecuteQuery" name="parameters" />
  </message>
  <message name="ExecuteQueryOut">
    <part element="tns:ExecuteQueryResponse" name="parameters" />
  </message>
  <portType name="queryDefMethodsSoap">
    <operation name="ExecuteQuery">
      <input message="tns:ExecuteQueryIn" />
      <output message="tns:ExecuteQueryOut" />
    </operation>
  </portType>
<binding name="queryDefMethodsSoap" type="tns:queryDefMethodsSoap">
  <soap:binding style="document" transport="http://schemas.xmlsoap.org/soap/http" />
  <operation name="ExecuteQuery">
    <soap:operation soapAction="xtk:queryDef#ExecuteQuery" style="document" />
    <input>
      <soap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" use="literal" />
    </input>
    <output>
      <soap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" use="literal" />
    </output>
  </operation>
    </binding>
    <service name="XtkQueryDef">
      <port binding="tns:queryDefMethodsSoap" name="queryDefMethodsSoap">
        <soap:address location="https://xxxxxxxxxxxxxx/nl/jsp/soaprouter.jsp" />
      </port>
    </service>
</definitions>

期望生成的SOAP请求Payload #

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:urn="urn:xtk:queryDef">
   <soapenv:Header/>
   <soapenv:Body>
      <urn:ExecuteQuery>
         <urn:sessiontoken>xxxxxxx</urn:sessiontoken>
         <urn:entity>
            <queryDef schema="nms:recipient" operation="select">
              <select>
                <node expr="@email"/>
                <node expr="@lastName+'-'+@firstName"/>
                <node expr="Year(@birthDate)"/>
              </select>
              <orderBy>
                <node expr="@birthDate" sortDesc="true"/>
              </orderBy>
        </queryDef>
         </urn:entity>
      </urn:ExecuteQuery>
   </soapenv:Body>
</soapenv:Envelope>

当前编写的Python代码 #

# Executes a query and returns the result set
def execute_query(session_token):
    # Load the WSDL locally - not authorised to get from server
    wsdl_url = os.path.abspath("querydef_dev.wsdl")
    history = HistoryPlugin()
    client = Client(wsdl_url, plugins=[history])

    execute_query_type = client.get_element("ns0:ExecuteQuery")

    entity_type = client.get_type("ns0:Element")
    any_entity = xsd.AnyObject(entity_type, entity_type(_value_1={'queryDef': [{'schema': 'recipients'}]}))

    params = execute_query_type(entity=any_entity, sessiontoken=session_token)

    response = client.service.ExecuteQuery(params)


if __name__ == '__main__':
    execute_query('xxxxxxx')

报错信息 #

AttributeError: 'dict' object has no attribute 'value'. Did you mean: 'values'?

我试过多种get_type、get_element的组合，也试过用**params和直接传params调用服务，都失败了。请问哪里出错了？

解决方案 #

问题出在对xsd.AnyObject的使用方式上，Zeep处理any类型时，不需要用entity_type去包裹字典，直接把对应XML结构的字典或者原生XML节点传给xsd.AnyObject即可。

修改后的代码：

from zeep import Client, xsd
from zeep.plugins import HistoryPlugin
import os

# 执行查询并返回结果集
def execute_query(session_token):
    # 本地加载WSDL - 无权限从服务器获取
    wsdl_url = os.path.abspath("querydef_dev.wsdl")
    history = HistoryPlugin()
    client = Client(wsdl_url, plugins=[history])

    # 构造对应期望XML的字典结构
    query_def = {
        'queryDef': {
            '@schema': 'nms:recipient',
            '@operation': 'select',
            'select': {
                'node': [
                    {'@expr': '@email'},
                    {'@expr': '@lastName+'-'+@firstName'},
                    {'@expr': 'Year(@birthDate)'}
                ]
            },
            'orderBy': {
                'node': {'@expr': '@birthDate', '@sortDesc': 'true'}
            }
        }
    }

    # 直接用xsd.AnyObject包装字典，无需用entity_type实例化
    any_entity = xsd.AnyObject(client.get_type("ns0:Element"), query_def)

    # 直接调用服务，不用提前构造execute_query_type实例，Zeep会自动映射参数
    response = client.service.ExecuteQuery(
        sessiontoken=session_token,
        entity=any_entity
    )

    # 可通过history查看生成的请求内容
    print(history.last_sent)
    return response


if __name__ == '__main__':
    execute_query('xxxxxxx')

关键修改点 #

• xsd.AnyObject正确用法：第二个参数直接传要序列化的字典（对应XML结构），不用entity_type(_value_1=...)包裹，因为entity_type实例化后是Zeep对象，而xsd.AnyObject需要原始数据结构或XML元素。
• 简化参数传递：不用先构造execute_query_type实例，直接给client.service.ExecuteQuery传命名参数，Zeep会根据WSDL自动生成请求结构。
• 字典匹配XML结构：用@前缀表示XML属性（比如@schema对应<queryDef schema="...">），嵌套字典对应子节点，列表对应重复子节点（如多个<node>）。

如果要更精准控制XML生成，也可以直接传lxml的Element对象给xsd.AnyObject：

from lxml import etree

query_def_elem = etree.fromstring("""
<queryDef schema="nms:recipient" operation="select">
  <select>
    <node expr="@email"/>
    <node expr="@lastName+'-'+@firstName"/>
    <node expr="Year(@birthDate)"/>
  </select>
  <orderBy>
    <node expr="@birthDate" sortDesc="true"/>
  </orderBy>
</queryDef>
""")
any_entity = xsd.AnyObject(client.get_type("ns0:Element"), query_def_elem)

内容的提问来源于stack exchange，提问作者mroshaw