0x02-用 Python 的 SOAP client Zeep 解析 Webservice 接口

python -mzeep "https://{{username}}:{{password}}@example?wsdl"

client = zeep.Client(
    wsdl=wsdl_url,
    transport=transport_with_basic_auth
)
client.wsdl.dump()

from collections import OrderedDict
from requests import Session
from requests.auth import HTTPBasicAuth
import zeep
from zeep.transports import Transport

def foo() -> OrderedDict:
    username = '{{USERNAME}}'
    password = '{{PASSWORD}}'
    wsdl_url = ('https://example.com/',
                'URI?wsdl')
    session = Session()
    session.auth = HTTPBasicAuth(username, password)
    transport_with_basic_auth = Transport(session=session)

    client = zeep.Client(
        wsdl=wsdl_url,
        transport=transport_with_basic_auth
    )

    # client.wsdl.dump()

    request_informations = client.get_type('ns0:foo')()
    request_requestinfo = client.get_type('ns0:bar')()

    foo_dict = {
        # ...
    }

    request_informations.__values__ = OrderedDict(request_dict)

    bar_dict = {
        # ...
    }

    request_requestinfo.__values__ = OrderedDict(bar_dict)

    resp_to_dict = OrderedDict()

    with transport_with_basic_auth.settings(timeout=2000):
        resp = client.service.execute(
            foo=request_informations, bar=request_requestinfo)

        resp_to_dict = resp.__dict__['__values__']

    return resp_to_dict

def use_esb_soap_use_case(
    url: str,
    username: str,
    password: str,
    case: dict
) -> OrderedDict:
    user = User()
    user.username = username
    user.password = password
    transport_with_basic_auth = auth.add_zeep_auth(
        user, context.CONTENT_TYPE_ZEEP_SOAP)
    client = Client(wsdl=url, transport=transport_with_basic_auth)

    resp_to_dict = OrderedDict()
    # case['some_filed'].update(field=value)

    with transport_with_basic_auth.settings(timeout=context.TIMEOUT_ZEEP_SOAP):
        try:
            resp = client.service.execute(**case)
            resp_to_dict = resp.__dict__['__values__']
        except ValidationError as validationError:
            logger.warn((validationError.message, "请检查 Body 是否有新增字段"))

    return resp_to_dict

factory = client.type_factory('ns0')
user = factory.User(name='John')
Return type: Factory

headers = {
    'Authorization': auth,
    'Content-Type': context.CONTENT_TYPE_APPLICATION_JSON
}
session = Session()
session.auth = HTTPBasicAuth(user.username, user.password)
session.headers.update(header)

from zeep import Client
from zeep import xsd

client = Client('http://my-entrprisy-endpoint.com')
client.service.submit_something(user_id=1, my_string=xsd.SkipValue)

Prefixes:
     xsd: http://www.w3.org/2001/XMLSchema
     ns0: urn:...

Global elements:

     ns0:Element_foo(attr1: ns0:{{element or type}})
     ...

Global types:
     xsd:anyType
     ...

Bindings:
     Soap11Binding: {urn:***}***
     Soap12Binding: {urn:***}***

Service: {{some_method}}
     Port: *** (Soap11Binding: {urn***}***)
         Operations:
            FOO -> BAR
     ...