Part 4: SOAP Client Development with zeep

Working with zeep

zeep is the most practical Python library for consuming SOAP services. It handles WSDL parsing, XML serialisation, type mapping, and response deserialisation automatically. I have used zeep to integrate with banking payment gateways, government identity verification APIs, and legacy ERP inventory feeds — all of which required SOAP.

This part covers everything needed to build a robust zeep client: loading WSDLs, calling operations, mapping complex responses, managing persistent sessions, caching WSDLs for production, and debugging when things go wrong.

Installing zeep

pip install zeep[xmlsec]   # xmlsec adds XML signature support for WS-Security (Part 5)

For projects that do not need WS-Security:

pip install zeep

Basic Client Setup

The minimal zeep client loads a WSDL and calls an operation:

# basic_client.py
from zeep import Client

client = Client("http://www.dneonline.com/calculator.asmx?WSDL")
result = client.service.Add(intA=15, intB=27)
print(result)   # 42

zeep reads the WSDL, understands the Add operation takes two integers, and handles all the XML construction and parsing.

For services where the endpoint URL is different from where the WSDL lives (common with staging/production environments), override the endpoint:

from zeep import Client
from zeep.transports import Transport

wsdl_url = "http://example.com/inventory/service?wsdl"

# Override the endpoint while keeping the original WSDL
client = Client(wsdl_url)
service = client.create_service(
    binding_name="{http://example.com/inventory}InventoryBinding",
    address="https://api-staging.example.com/inventory",
)

result = service.GetProduct(ProductId="SKU-001")

create_service is the correct way to target a different endpoint than the WSDL specifies — I use this pattern in almost every real-world integration where there is a test environment URL separate from the WSDL URL.

Inspecting Available Operations

Before writing client code, I always check what operations are available:

# inspect_client.py
from zeep import Client

client = Client("http://localhost:8000/?wsdl")

# List all available operations
print(client.wsdl.dump())

This prints a human-readable summary of services, ports, and operations with their input/output types. Extremely useful when working with undocumented or poorly documented services.

Calling Operations and Handling Responses

Simple Types

# simple_call.py
from zeep import Client

client = Client("http://localhost:8000/?wsdl")

product = client.service.GetProduct(ProductId="SKU-001")

# zeep returns a zeep.objects type — access it like an object
print(product.ProductId)    # SKU-001
print(product.Name)         # Widget Pro
print(product.Price)        # Decimal('29.99')
print(product.InStock)      # True
print(product.StockCount)   # 150

Complex Nested Types

When an operation takes a complex type as input, pass it as a dictionary or construct it with the factory:

# complex_input.py
from decimal import Decimal
from zeep import Client

client = Client("http://localhost:8000/?wsdl")

# Method 1: Pass a plain dictionary (zeep converts it to the right type)
new_product = client.service.CreateProduct(
    product_input={
        "ProductId": "SKU-003",
        "Name": "Super Widget",
        "Price": Decimal("49.99"),
        "InStock": True,
    }
)

# Method 2: Use the zeep type factory for explicit typing
ProductInput = client.get_type("ns0:ProductInput")
input_obj = ProductInput(
    ProductId="SKU-004",
    Name="Mega Gadget",
    Price=Decimal("99.00"),
    InStock=False,
)
new_product = client.service.CreateProduct(product_input=input_obj)

I prefer Method 1 (plain dict) for simple cases and Method 2 (factory) when I need to reuse the type object, validate its shape, or pass it to multiple operations.

List Responses

When a response contains an array type, zeep returns it as a Python list:

# list_response.py
from zeep import Client
from zeep.helpers import serialize_object

client = Client("http://localhost:8000/?wsdl")

result = client.service.ListProducts(page=1, page_size=10)

print(f"Total: {result.TotalCount}")
print(f"Page:  {result.Page}")

for product in result.Products:
    print(f"  {product.ProductId}: {product.Name} ({product.Price})")

Serialising Responses to Dictionaries

zeep's native objects can be converted to plain Python dicts using serialize_object:

# serialize_response.py
from zeep import Client
from zeep.helpers import serialize_object

client = Client("http://localhost:8000/?wsdl")
product = client.service.GetProduct(ProductId="SKU-001")

# Convert to dict (useful for JSON serialisation, logging, testing)
product_dict = serialize_object(product)
print(product_dict)
# {
#   'ProductId': 'SKU-001',
#   'Name': 'Widget Pro',
#   'Price': Decimal('29.99'),
#   'InStock': True,
#   'StockCount': 150
# }

# Convert to JSON-safe dict  
import json
from decimal import Decimal

def decimal_default(obj):
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError

product_json = json.dumps(product_dict, default=decimal_default)

Session Management and Persistent Connections

SOAP services over HTTPS benefit from HTTP keep-alive and session cookie support. Use requests.Session with zeep's Transport:

# session_client.py
import requests
from zeep import Client
from zeep.transports import Transport

session = requests.Session()

# If the service uses basic HTTP auth (not WS-Security)
session.auth = ("api_username", "api_password")

# Add custom headers required by the service
session.headers.update({
    "X-Client-ID": "my-integration-client",
    "X-Request-Source": "python-client",
})

# Configure timeouts: (connect_timeout, read_timeout)
transport = Transport(
    session=session,
    timeout=30,
    operation_timeout=60,
)

client = Client(
    wsdl="https://api.example.com/service?wsdl",
    transport=transport,
)

# All requests from this client use the same session
result = client.service.GetProduct(ProductId="SKU-001")

For production clients, always set explicit timeouts. SOAP services — especially older enterprise systems — can hang indefinitely on network issues. A client without timeouts will block threads and exhaust thread pools.

Caching WSDLs for Production

Loading a WSDL on every client instantiation means an HTTP request to the service. For high-volume applications or services with slow WSDL endpoints, cache the WSDL locally:

# cached_client.py
import os
import tempfile
from zeep import Client
from zeep.cache import SqliteCache
from zeep.transports import Transport

# zeep's built-in SQLite cache (timeout in seconds; 3600 = 1 hour)
cache = SqliteCache(
    path=os.path.join(tempfile.gettempdir(), "zeep_wsdl_cache.db"),
    timeout=3600,
)

transport = Transport(cache=cache)

client = Client(
    wsdl="https://api.example.com/service?wsdl",
    transport=transport,
)

With SqliteCache, the first request downloads and caches the WSDL. Subsequent instantiations resolve from the cache until the timeout expires.

For long-running services or when you know the WSDL will never change, download it once and load from disk:

# local_wsdl_client.py
import os
from pathlib import Path
from zeep import Client

wsdl_path = Path("wsdl/inventory-service.wsdl")

if not wsdl_path.exists():
    # Download once and save
    import requests
    wsdl_path.parent.mkdir(parents=True, exist_ok=True)
    resp = requests.get("https://api.example.com/service?wsdl", timeout=30)
    resp.raise_for_status()
    wsdl_path.write_text(resp.text, encoding="utf-8")

# Load from local file — no network roundtrip
client = Client(wsdl=f"file://{wsdl_path.absolute()}")
service = client.create_service(
    binding_name="{http://example.com/inventory}InventoryBinding",
    address="https://api.example.com/service",
)

Debugging: Logging Raw SOAP Traffic

When the client behaves unexpectedly, the first thing I do is enable transport logging to see the raw XML on the wire.

# debug_client.py
import logging

# Enable zeep's transport logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("zeep.transports").setLevel(logging.DEBUG)

from zeep import Client

client = Client("http://localhost:8000/?wsdl")
client.service.GetProduct(ProductId="SKU-001")

The log output shows:

• The full HTTP request with headers and body

• The raw SOAP response XML

• Any XML parsing or binding steps

For production, log only errors:

logging.getLogger("zeep.transports").setLevel(logging.ERROR)

To capture the raw XML programmatically without printing to logs, use a custom transport:

# capture_raw_xml.py
import io
from zeep import Client
from zeep.transports import Transport
from requests import Session


class CapturingTransport(Transport):
    """Transport that captures the last request and response XML."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.last_request: bytes | None = None
        self.last_response: bytes | None = None

    def post_xml(self, address, envelope, headers):
        from lxml import etree
        self.last_request = etree.tostring(envelope, pretty_print=True)
        response = super().post_xml(address, envelope, headers)
        self.last_response = response.content
        return response


transport = CapturingTransport()
client = Client("http://localhost:8000/?wsdl", transport=transport)

client.service.GetProduct(ProductId="SKU-001")

print("=== REQUEST ===")
print(transport.last_request.decode())

print("=== RESPONSE ===")
print(transport.last_response.decode())

Handling Optional Fields and None Values

SOAP types have minOccurs="0" fields that behave differently from what Python developers expect. zeep returns None for absent optional elements:

# optional_fields.py
from zeep import Client

client = Client("http://localhost:8000/?wsdl")
product = client.service.GetProduct(ProductId="SKU-001")

# Always guard optional fields
in_stock = product.InStock if product.InStock is not None else False
stock_count = product.StockCount if product.StockCount is not None else 0

When building input for operations with many optional fields, only include the ones you need in the dict:

# Omitting optional fields — just do not include them in the dict
result = client.service.CreateProduct(
    product_input={
        "ProductId": "SKU-005",
        "Name": "Basic Item",
        "Price": "9.99",
        # InStock is optional; omitting it — service will use its default
    }
)

Building a Production-Ready Client Class

For real projects, I wrap the zeep client in a class that manages its lifecycle, adds retry logic, and provides typed methods:

# inventory_client.py
from __future__ import annotations

import logging
import time
from decimal import Decimal
from functools import wraps
from typing import Optional

import requests
from zeep import Client
from zeep.cache import SqliteCache
from zeep.exceptions import Fault, TransportError
from zeep.helpers import serialize_object
from zeep.transports import Transport

logger = logging.getLogger(__name__)


def _retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry decorator for transient network errors."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except TransportError as exc:
                    last_exc = exc
                    if attempt < max_attempts:
                        logger.warning(
                            "Transport error (attempt %d/%d): %s",
                            attempt, max_attempts, exc,
                        )
                        time.sleep(delay * attempt)
                except Fault:
                    raise   # SOAP Faults are not retryable
            raise last_exc
        return wrapper
    return decorator


class InventoryClient:
    """
    Typed client for the Inventory SOAP service.

    Usage:
        client = InventoryClient("https://api.example.com/inventory?wsdl")
        product = client.get_product("SKU-001")
    """

    def __init__(
        self,
        wsdl_url: str,
        endpoint_url: str | None = None,
        timeout: int = 30,
        operation_timeout: int = 60,
        wsdl_cache_ttl: int = 3600,
    ):
        session = requests.Session()
        session.timeout = (timeout, operation_timeout)

        cache = SqliteCache(timeout=wsdl_cache_ttl)
        transport = Transport(session=session, cache=cache)

        raw_client = Client(wsdl=wsdl_url, transport=transport)

        if endpoint_url:
            # Determine binding name from the WSDL
            services = list(raw_client.wsdl.services.values())
            ports = list(services[0].ports.values())
            binding_name = str(ports[0].binding.name)

            self._service = raw_client.create_service(
                binding_name=binding_name,
                address=endpoint_url,
            )
        else:
            self._service = raw_client.service

    @_retry(max_attempts=3)
    def get_product(self, product_id: str) -> dict:
        """Return product data as a plain dict."""
        result = self._service.GetProduct(ProductId=product_id)
        return serialize_object(result)

    @_retry(max_attempts=3)
    def list_products(self, page: int = 1, page_size: int = 10) -> dict:
        """Return paginated product list."""
        result = self._service.ListProducts(page=page, page_size=page_size)
        return serialize_object(result)

    @_retry(max_attempts=2)
    def update_stock(self, product_id: str, new_count: int) -> dict:
        """Update stock count for a product."""
        result = self._service.UpdateStock(
            product_id=product_id,
            new_stock_count=new_count,
        )
        return serialize_object(result)

Usage:

# main.py
from inventory_client import InventoryClient
from zeep.exceptions import Fault

client = InventoryClient(
    wsdl_url="https://api.example.com/inventory?wsdl",
    endpoint_url="https://api-prod.example.com/inventory",
    timeout=10,
    operation_timeout=30,
)

try:
    product = client.get_product("SKU-001")
    print(f"{product['Name']}: {product['Price']}")
except Fault as fault:
    print(f"Service error: {fault.message}")

What's Next

You can now build robust zeep clients that:

• Load and cache WSDLs

• Call operations with simple and complex types

• Debug raw SOAP traffic

• Manage persistent sessions with timeouts

• Retry on transient network failures

In Part 5, we add security to SOAP messages using WS-Security — covering UsernameToken, message timestamps, and digital signatures with X.509 certificates.