Zeep

lastmod 2024-10-28created 2021-04-0827 edits

A fast and modern Python SOAP client

Highlights

Compatible with Python 3.6, 3.7, 3.8 and PyPy
Build on top of lxml and requests
Support for Soap 1.1, Soap 1.2 and HTTP bindings
Support for WS-Addressing headers
Support for WSSE (UserNameToken / x.509 signing)
Support for asyncio via httpx
Experimental support for XOP messages

Installation

pip install zeep

Client.create_service

Binding Nmae (QName)

The first argument is the binding name.

The binding name belongs to the XML Namespace "wsdl" (whose address is http://schemas.xmlsoap.org/wsdl/) [^0]

<wsdl:definitions xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/" ...>

Check the "name" attribute of the "wsdl:binding" tag for the binding name:[^1]

<wsdl:binding name="???" ...>

The final QName to pass is the binding name appended to the WSDL Declaration location.

For example ...

WSDL Declaration location: http://www.onvif.org/ver10/device/wsdl
Binding name: <wsdl:binding name="DeviceBinding" ...>

In this case, the binding string to pass as the argument is "{http://www.onvif.org/ver10/device/wsdl}DeviceBinding".

File Cache

For this, the client constructor downloads schema files. This step takes a long time. So these files can be cached, but there turned out to be only a memory cache and an sqlite cache.. So I add a file cache as well:

import os
from base64 import b64encode
from typing import Final, Optional, Any
from logging import getLogger
from zeep import Client
from zeep.cache import Base
from zeep.transports import Transport
from zeep.wsse.username import UsernameToken

logger = getLogger()

CURRENT_DIR: Final[str] = os.path.dirname(__file__)
CACHE_DIR: Final[str] = os.path.join(CURRENT_DIR, "cache")
DEFAULT_ENCODING: Final[str] = "latin1"

ONVIF_DEVICE_WSDL_NAMESPACE = "http://www.onvif.org/ver10/device/wsdl"
ONVIF_DEVICE_WSDL_PATH = os.path.join(
    ONVIF_DIR, "ver10", "device", "wsdl", "devicemgmt.wsdl"
)
ONVIF_DEVICE_BINDING_QNAME = f"{{{ONVIF_DEVICE_WSDL_NAMESPACE}}}DeviceBinding"

class FileCache(Base):
    def add(self, url: str, content: Any):
        b64_name = b64encode(url.encode(encoding=DEFAULT_ENCODING))
        filename = str(b64_name, encoding=DEFAULT_ENCODING)
        filepath = os.path.join(CACHE_DIR, filename)
        try:
            if not os.path.exists(filepath):
                with open(filepath, "wb") as f:
                    f.write(content)
        except BaseException as e:  # noqa
            logger.error(f"FileCache.add(url={url}) error: {e}")

    def get(self, url: str):
        b64_name = b64encode(url.encode(encoding=DEFAULT_ENCODING))
        filename = str(b64_name, encoding=DEFAULT_ENCODING)
        filepath = os.path.join(CACHE_DIR, filename)
        try:
            if os.path.isfile(filepath):
                with open(filepath, "rb") as f:
                    return f.read()
        except BaseException as e:  # noqa
            logger.error(f"FileCache.get(url={url}) error: {e}")
        return None

def get_onvif_device_capabilities(
    address: str,
    wsse: Optional[UsernameToken] = None,
) -> Any:
    transport = Transport(cache=FileCache())
    client = Client(wsdl=ONVIF_DEVICE_WSDL_PATH, wsse=wsse, transport=transport)
    service = client.create_service(ONVIF_DEVICE_BINDING_QNAME, address)
    return service.GetCapabilities()

A simple example

from zeep import Client

client = Client('http://www.webservicex.net/ConvertSpeed.asmx?WSDL')
result = client.service.ConvertSpeed(
    100, 'kilometersPerhour', 'milesPerhour')

assert result == 62.137

How to include HTTP headers

Including them in Settings:

import zeep

settings = zeep.Settings(extra_http_headers={'Authorization': 'Bearer ' + token})
client = zeep.Client(wsdl=url, settings=settings)

Including them in Transport:

import requests
from zeep import Client, Transport

headers = {
    "Authorization": "Bearer " + get_token()
}
session = requests.Session()
session.headers.update(headers)
transport = Transport(session=session)
client = Client(wsdl=url, transport=transport)

Applying them after object initialization:

client.transport.session.headers.update({'yourHeader': 'yourValue'})

HTTP Authentication

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))

See HTTPBasicAuth and HTTPDigestAuth for details.

WS-Security (WSSE)

WS-Security (WSSE) — Zeep 4.1.0 documentation

UsernameToken

from zeep import Client
from zeep.wsse.username import UsernameToken
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    wsse=UsernameToken('username', 'password'))

Signature (x509)

To use the wsse.Signature() plugin you will need to install the xmlsec module. See the README for xmlsec for the required dependencies on your platform.

To append the security token as BinarySecurityToken, you can use wsse.BinarySignature() plugin.

Example usage A:

from zeep import Client
from zeep.wsse.signature import Signature
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    wsse=Signature(
        private_key_filename, public_key_filename,
        optional_password))

Example usage B:

from zeep import Client
from zeep.wsse.signature import Signature
from zeep.transports import Transport
from requests import Session
session = Session()
session.cert = '/path/to/ssl.pem'
transport = Transport(session=session)
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    transport=transport)

UsernameToken and Signature together

To use UsernameToken and Signature together, then you can pass both together to the client in a list

from zeep import Client
from zeep.wsse.username import UsernameToken
from zeep.wsse.signature import Signature
user_name_token = UsernameToken('username', 'password')
signature = Signature(private_key_filename, public_key_filename,
    optional_password)
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    wsse=[user_name_token, signature])

ONVIF Example

from zeep import Client
from zeep.wsse.username import UsernameToken

wsdl = "http://www.onvif.org/ver10/media/wsdl/media.wsdl"
client = Client(wsdl=wsdl, wsse=UsernameToken("admin", "0000", use_digest=True))
service = client.create_service("{http://www.onvif.org/ver10/media/wsdl}MediaBinding", "http://192.168.0.50/onvif/device_service")
profiles = service.GetProfiles()
print(profiles)

factory = client.type_factory("http://www.onvif.org/ver10/schema")
transport = factory.Transport(Protocol='RTSP')
setup = factory.StreamSetup(Stream='RTP-Unicast', Transport=transport)
result = service.GetStreamUri(StreamSetup=setup, ProfileToken='Profile1')
print(result)

Json Serialize

from json import dumps
from zeep import helpers

o = helpers.serialize_object(zeep_object, dict)
json_text = dumps(o)
print(json_text)

Problems can occur with the datetime and timedelta types. Dump with the following function.

from datetime import datetime, timedelta
from json import JSONEncoder, dumps

def dumps_default(o: Any) -> Any:
    if isinstance(o, datetime):
        return o.isoformat()
    elif isinstance(o, timedelta):
        return o.total_seconds()
    try:
        return JSONEncoder().default(o)
    except TypeError:
        return str(o)

# ...

json = dumps(o, indent=4, sort_keys=True, default=dumps_default)

simpleType restriction

Zeep dont handle restriction on simpleTypes? · Issue #513 · mvantellingen/python-zeep

For reading things like type restrictions in the xds schema files of a wsdl. For example, content like the following:

<xs:simpleType name="RollenEnum">
    <xs:restriction base="xs:string">
        <xs:enumeration value="VersichertePerson"/>
        <xs:enumeration value="Versicherungsnehmer"/>
        <xs:enumeration value="Beitragszahler"/>
        <xs:enumeration value="BezugsberechtigterImTodesfall"/>
        <xs:enumeration value="Ehepartner"/>
        <xs:enumeration value="GesetzlicherVertreter"/>
    </xs:restriction>
</xs:simpleType>

zeep's SchemaVisitor turned out to have no implementation for reading the above... So it cannot be used as of now (2024-10-25).

Favorite site

author Zer0 sitemap