Zeep

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)

첫 번째 인자는 바인딩 이름을 쓰면 된다.

바인딩 이름은 "wsdl" 인 XML Namespace (주소는 http://schemas.xmlsoap.org/wsdl/ 이다) [^0]

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

바인딩 이름은 "wsdl:binding" Tag 의 "name" 속성(Attribute)을 확인하면 된다:[^1]

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

최종적으로 입력할 QName은 WSDL Declaration 위치에 바인딩 이름을 추가하면 된다.

예를 들어 ...

WSDL Declaration 위치: http://www.onvif.org/ver10/device/wsdl
바인딩 이름: <wsdl:binding name="DeviceBinding" ...>

일 경우 인자로 전달할 바인딩 문자열은 "{http://www.onvif.org/ver10/device/wsdl}DeviceBinding" 이다.

File Cache

위해 클라이언트 생성자에서 스키마 파일을 다운로드 받는다. 이 과정의 시간이 오래걸린다. 따라서 해당 파일들은 캐시하도록 할 수 있는데 메모리 캐시랑 sqlite 캐시만 있더라.. 그래서 파일 캐시도 추가한다:

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

HTTP Header 포함 방법

Settings 에 포함하는 방법:

import zeep

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

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)

객체 초기화 이 후 적용 방법:

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

자세한 내용은 HTTPBasicAuth, HTTPDigestAuth 항목들 참조.

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)

datetime 와 timedelta 타입에 문제가 발생될 수 있다. 다음 함수로 dump 하자.

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

wsdl의 xds 스키마 파일의 타입 제약사항같은걸 읽기위함. 예를 들면 다음과 같은 내용:

<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의 SchemaVisitor에 위 내용은 읽는 구현이 없더라... 그래서 지금 현재(2024-10-25) 사용할 수 없다.