DOCSEARCHbyAlgolia
Submit Docs

zeep Python SOAP Client

repository·main·Indexed 24 days ago

https://github.com/mvantellingen/python-zeep

A Python SOAP client designed to interact with SOAP web services. Built on lxml, requests, and httpx, zeep provides support for various SOAP versions and security standards. It features a Client class for WSDL-based interaction, AsyncClient for asynchronous operations, and tools for handling complex XSD types, multipart attachments, and custom SOAP headers.

Tokens
14.7K
Snippets
47
Records
96
Agent score
83%

What's inside zeep

  1. Write a custom Zeep plugin

    main

    You can create custom plugins to process or modify data during the SOAP lifecycle. Plugins can intercept data at two stages:

    1. ingress: Triggered after a response is received from the server but before it is processed by the client. The envelope is the full SOAP envelope.
    2. egress: Triggered before a request is sent to the server. The envelope contains only the body of the SOAP message.

    Both methods must return a tuple containing the envelope (an lxml element) and the http_headers.

    from lxml import etree
from zeep import Plugin

class MyLoggingPlugin(Plugin):

    def ingress(self, envelope, http_headers, operation):
        # Process received response
        print(etree.tostring(envelope, pretty_print=True))
        return envelope, http_headers

    def egress(self, envelope, http_headers, operation, binding_options):
        # Process outgoing request
        print(etree.tostring(envelope, pretty_print=True))
        return envelope, http_headers

  2. Get raw HTTP responses from Zeep service calls

    main

    By default, Zeep processes SOAP responses into Python objects. To bypass this and receive the underlying requests.Response object directly, set the raw_response setting to True. This can be done via the client's settings attribute or within a with client.settings(...) context manager block.

    with client.settings(raw_response=True):
    response = client.service.myoperation()
    # response is a requests.Response object

  3. Disable Strict mode for non-compliant SOAP servers

    main

    By default, Zeep operates in 'strict' mode. If you are working with a SOAP server that does not strictly follow standards, you can set strict=False in your Settings.

    Warning: Disabling strict mode is a last resort as it enables XML recovery mode and allows missing non-optional elements in xsd:sequences, which may lead to data loss during XML/response conversion.

    from zeep import Client, Settings

settings = Settings(strict=False)
client = Client('http://my-wsdl/wsdl', settings=settings)

  4. Handle xsd:choice elements

    main

    Mapping xsd:choice elements to code is handled via two primary methods:

    1. Simple choices: For simple definitions, you can pass the elements of the choice as keyword arguments (kwargs) directly to the element constructor.

      element = client.get_element('ns0:ElementName')
obj = element(item_1='foo')

    2. Complex or Nested choices: For more complex structures, use the special keyword argument _value_N, where N is the index of the choice in the parent type.

      • For maxOccurs="1", pass a dictionary: element(_value_1={'item_1_a': 'foo', 'item_1_b': 'bar'}).
      • For maxOccurs="unbounded", pass a list of dictionaries: element(_value_1=[{'item_1': 'foo'}, {'item_2': 'bar'}]).
    # Simple choice
element = client.get_element('ns0:ElementName')
obj = element(item_1='foo')

# Nested choice (maxOccurs=1)
element = client.get_element('ns0:ElementName')
obj = element(_value_1={'item_1_a': 'foo', 'item_1_b': 'bar'})

# Nested list choice (maxOccurs=unbounded)
element = client.get_element('ns0:ElementName')
obj = element(_value_1=[{'item_1': 'foo'}, {'item_2': 'bar'}])

  5. Enable WS-Addressing (WSA) support

    main

    Zeep provides experimental support for the WS-Addressing specification, which uses soap:Header elements to enable advanced routing of SOAP messages.

    If the WSDL document explicitly defines that WSA is required, Zeep will automatically include the necessary headers. If you need to manually enable or customize WSA behavior, you must add the WsAddressingPlugin to the Client.plugins list during client initialization.

    from zeep import Client
from zeep.wsa import WsAddressingPlugin

client = Client(
    'http://examples.python-zeep.org/basic.wsdl',
    plugins=[WsAddressingPlugin()]
)
client.service.DoSomething()

  7. Configure Async HTTP Authentication

    main

    The zeep.AsyncClient uses a different backend (httpx). To use authentication with the async client, create an httpx.AsyncClient with the required auth and pass it to zeep.transports.AsyncTransport.

    import httpx
import zeep
from zeep.transports import AsyncTransport

USER = 'username'
PASSWORD = 'password'

httpx_client = httpx.AsyncClient(auth=(USER, PASSWORD))

aclient = zeep.AsyncClient(
    "http://my-endpoint.com/production.svc?wsdl",
    transport=AsyncTransport(client=httpx_client)
)

  8. Configure TLS verification with Transport

    main

    To verify TLS connections (e.g., when using self-signed certificates), create a requests.Session instance, set the verify attribute to the path of your CA bundle (an X.509 ASCII .pem or .crt file containing both root and intermediate CAs), and pass this session to a zeep.transports.Transport instance.

    Alternatively, you can use session.cert to provide a TLS client certificate.

    To disable TLS verification (not recommended): Set session.verify = False. This should only be used for testing. Python's urllib3 will emit an InsecureRequestWarning.

    from requests import Session
from zeep import Client
from zeep.transports import Transport

session = Session()
session.verify = 'path/to/my/certificate.pem'
transport = Transport(session=session)
client = Client(
    'http://my.own.sslhost.local/service?WSDL',
    transport=transport)

  9. Configure HTTP Authentication

    main

    If the SOAP service requires HTTP Authentication (rather than security headers inside the SOAP message), create a requests.Session object, set its auth attribute using a requests.auth object (like HTTPBasicAuth, HTTPDigestAuth, or OAuth1), and pass that session to the Transport class.

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

  10. Handle SOAP multipart attachments

    main

    When a SOAP server responds with a Content-type: multipart header, Zeep returns a MessagePack object instead of a standard response object. This object contains a root attribute representing the main SOAP body and an attachments list containing the multipart data.

    You can access attachments in two ways:

    1. By index: Access elements in the attachments list directly.
    2. By content_id: Use the get_by_content_id(content_id) method on the MessagePack object to retrieve a specific attachment using its unique identifier.

    Each attachment object has a .content attribute which contains the raw bytes of the attachment.

    from zeep import Client

client = Client('http://www.risky-stuff.com/claim.svc?wsdl')

# The returned object is a MessagePack object due to multipart response
pack = client.service.GetClaimDetails('061400a')

# Access the main SOAP body
ClaimDetails = pack.root

# Access attachments by index
SignedFormTiffImage = pack.attachments[0].content
CrashPhotoJpeg = pack.attachments[1].content

# Or lookup by content_id
attachment = pack.get_by_content_id('<claim061400a.tiff@claiming-it.com>').content

  11. Use caching backends with Transport

    main

    For improved performance, you can use a caching backend to store WSDL and XSD files. By default, Zeep does not use caching.

    Available backends:

    • SqliteCache: Recommended for performance. It caches files for 1 hour by default. You can customize the database path and timeout.
    • InMemoryCache: Uses a global dictionary to store URLs and their content.
    from zeep import Client
from zeep.cache import SqliteCache
from zeep.transports import Transport

transport = Transport(cache=SqliteCache())
client = Client(
    'http://www.webservicex.net/ConvertSpeed.asmx?WSDL',
    transport=transport)

  12. Configure Zeep settings using a context manager

    main

    You can temporarily modify client settings using a context manager. This is useful for scoped changes, such as instructing Zeep to return the raw HTTP response instead of the processed SOAP response. When using client.settings(raw_response=True), the service call will return a requests.Response object.

    from zeep import Client
from zeep import xsd

client = Client('http://my-endpoint.com/production.svc?wsdl')

with client.settings(raw_response=True):
    response = client.service.myoperation()

    # response is now a regular requests.Response object
    assert response.status_code == 200
    assert response.content