如何获取Spyne服务中SOAP Action名称列表?
获取Spyne服务的SOAP Action列表方案
你可以通过Spyne的Application实例直接提取SOAP Action,完全不需要硬编码;也可以从XML Schema中解析,但前者更直接高效。以下是具体实现方案:
方法1:从Application的服务方法中提取(推荐)
Spyne的Application实例会维护所有注册的服务及方法,每个方法的function属性中存储了定义时指定的_soap_action值。直接遍历这些方法就能拿到所有可用的SOAP Action:
from spyne import Application, rpc, ServiceBase, Integer, Unicode from spyne.protocol.soap import Soap11 # 示例服务定义 class ServiceA(ServiceBase): @rpc(Integer, Integer, _returns=Integer, _soap_action='http://example.com/ServiceA/Add') def add(ctx, a, b): return a + b class ServiceB(ServiceBase): @rpc(Unicode, _returns=Unicode, _soap_action='http://example.com/ServiceB/Echo') def echo(ctx, msg): return msg # 创建Application实例 app = Application([ServiceA, ServiceB], tns='http://example.com/soap', in_protocol=Soap11(validator='lxml'), out_protocol=Soap11()) # 提取所有SOAP Action soap_actions = [] for service in app.services: for method in service.methods.values(): action = method.function.soap_action if action: # 过滤未显式指定SOAP Action的方法 soap_actions.append(action) print(soap_actions) # 输出: ['http://example.com/ServiceA/Add', 'http://example.com/ServiceB/Echo']
这个方案完全依赖服务定义本身,只要你在@rpc装饰器中正确设置了_soap_action,就能自动同步获取最新的Action列表,无需手动维护。
方法2:从XML Schema中提取(备选)
如果一定要通过xml_schema获取,可以解析Application生成的WSDL文档,从中提取operation对应的soapAction属性。这种方法需要处理XML命名空间和节点结构,适合需要基于Schema做额外处理的场景:
from lxml import etree # 生成并解析WSDL的XML树 schema_doc = app.interface.xml_schema_doc() root = etree.fromstring(schema_doc) # 定义WSDL和SOAP的命名空间 ns_map = { 'wsdl': 'http://schemas.xmlsoap.org/wsdl/', 'soap': 'http://schemas.xmlsoap.org/wsdl/soap/' } # 提取所有soapAction值 soap_actions_from_schema = [] for operation in root.xpath('//wsdl:operation', namespaces=ns_map): action_attr = operation.xpath('./soap:operation/@soapAction', namespaces=ns_map) if action_attr: soap_actions_from_schema.append(action_attr[0]) print(soap_actions_from_schema) # 输出与方法1一致
路由场景的实用建议
在路由逻辑中,你可以将提取到的SOAP Action与对应的Application实例建立映射,实现动态路由:
# 假设有两个独立的Application实例对应不同服务 app_a = Application([ServiceA], tns='http://example.com/soap/a', in_protocol=Soap11(), out_protocol=Soap11()) app_b = Application([ServiceB], tns='http://example.com/soap/b', in_protocol=Soap11(), out_protocol=Soap11()) # 构建Action到Application的映射 action_app_map = { 'http://example.com/ServiceA/Add': app_a, 'http://example.com/ServiceB/Echo': app_b } # 路由中间件示例(WSGI环境) def soap_router(environ, start_response): # 从请求头中获取SOAP Action(注意去除首尾的引号) soap_action = environ.get('HTTP_SOAPACTION', '').strip('"') target_app = action_app_map.get(soap_action) if target_app: return target_app(environ, start_response) # 处理未匹配的情况 start_response('404 Not Found', [('Content-Type', 'text/plain')]) return [b'Specified SOAP Action is not supported']
内容的提问来源于stack exchange,提问作者John Gedik
相关产品推荐
相关产品推荐

