tessl/pypi-zeep

A comprehensive Python SOAP client library for consuming SOAP web services with support for SOAP 1.1/1.2, WSSE authentication, and async operations

Overview

Eval results

Files

Plugin System

Name: tessl/pypi-zeep
Author: tessl

Extensible plugin architecture for intercepting and modifying SOAP requests and responses. Plugins enable custom logging, authentication, caching, debugging, and other cross-cutting concerns without modifying core zeep behavior.

Capabilities

Base Plugin Class

Foundation for all zeep plugins with request/response interception points.

class Plugin:
    def ingress(self, envelope, http_headers: dict, operation):
        """
        Process incoming SOAP responses.
        
        Called when receiving responses from SOAP services.
        
        Parameters:
        - envelope: Response envelope (lxml.etree._Element)
        - http_headers: HTTP response headers dict
        - operation: Operation instance that generated the request
        
        Returns:
        tuple: (modified_envelope, modified_headers) or None to keep unchanged
        """

    def egress(self, envelope, http_headers: dict, operation, binding_options: dict):
        """
        Process outgoing SOAP requests.
        
        Called before sending requests to SOAP services.
        
        Parameters:
        - envelope: Request envelope (lxml.etree._Element)  
        - http_headers: HTTP request headers dict
        - operation: Operation instance being called
        - binding_options: Binding-specific options dict
        
        Returns:
        tuple: (modified_envelope, modified_headers) or None to keep unchanged
        """

History Plugin

Built-in plugin for tracking request/response history for debugging and auditing.

class HistoryPlugin(Plugin):
    def __init__(self, maxlen: int = 1):
        """
        Create history tracking plugin.
        
        Parameters:
        - maxlen: Maximum number of request/response pairs to store
        """

    @property
    def last_sent(self):
        """Last sent request envelope."""

    @property  
    def last_received(self):
        """Last received response envelope."""

Plugin Application Functions

Utility functions for applying plugins to requests and responses.

def apply_egress(client, envelope, http_headers: dict, operation, binding_options: dict):
    """
    Apply all egress plugins to outgoing request.
    
    Parameters:
    - client: Client instance with plugins
    - envelope: Request envelope
    - http_headers: HTTP headers
    - operation: Operation being called
    - binding_options: Binding options
    
    Returns:
    tuple: (processed_envelope, processed_headers)
    """

def apply_ingress(client, envelope, http_headers: dict, operation):
    """
    Apply all ingress plugins to incoming response.
    
    Parameters:
    - client: Client instance with plugins
    - envelope: Response envelope  
    - http_headers: HTTP headers
    - operation: Operation that generated request
    
    Returns:
    tuple: (processed_envelope, processed_headers)
    """

Usage Examples

Basic Plugin Usage

from zeep import Client
from zeep.plugins import HistoryPlugin

# Add history plugin to track requests/responses
history = HistoryPlugin()
client = Client('http://example.com/service.wsdl', plugins=[history])

# Make SOAP call
result = client.service.SomeOperation(param='value')

# Inspect request/response history
print("Last sent request:")
print(history.last_sent)

print("Last received response:")  
print(history.last_received)

Custom Logging Plugin

import logging
from zeep import Client
from zeep.plugins import Plugin

class LoggingPlugin(Plugin):
    def __init__(self, logger_name='zeep.plugin'):
        self.logger = logging.getLogger(logger_name)
    
    def egress(self, envelope, http_headers, operation, binding_options):
        self.logger.info(f"Sending request to operation: {operation.name}")
        self.logger.debug(f"Request envelope: {envelope}")
        return envelope, http_headers
    
    def ingress(self, envelope, http_headers, operation):
        self.logger.info(f"Received response from operation: {operation.name}")
        self.logger.debug(f"Response envelope: {envelope}")
        return envelope, http_headers

# Set up logging
logging.basicConfig(level=logging.INFO)

# Use custom plugin
logging_plugin = LoggingPlugin()
client = Client('http://example.com/service.wsdl', plugins=[logging_plugin])

result = client.service.SomeOperation(param='value')

Authentication Plugin

from zeep import Client
from zeep.plugins import Plugin
import base64

class CustomAuthPlugin(Plugin):
    def __init__(self, api_key):
        self.api_key = api_key
    
    def egress(self, envelope, http_headers, operation, binding_options):
        # Add custom authentication header
        http_headers['X-API-Key'] = self.api_key
        http_headers['Authorization'] = f'Bearer {self.api_key}'
        return envelope, http_headers

# Use authentication plugin
auth_plugin = CustomAuthPlugin('your-api-key-here')
client = Client('http://example.com/service.wsdl', plugins=[auth_plugin])

result = client.service.AuthenticatedOperation(param='value')

Request/Response Modification Plugin

from zeep import Client
from zeep.plugins import Plugin
from lxml import etree

class RequestModifierPlugin(Plugin):
    def egress(self, envelope, http_headers, operation, binding_options):
        # Add custom SOAP header
        header = envelope.find('.//{http://schemas.xmlsoap.org/soap/envelope/}Header')
        if header is None:
            header = etree.Element('{http://schemas.xmlsoap.org/soap/envelope/}Header')
            envelope.insert(0, header)
        
        # Add custom header element
        custom_header = etree.SubElement(header, '{http://example.com}CustomHeader')
        custom_header.text = 'Custom Value'
        
        return envelope, http_headers
    
    def ingress(self, envelope, http_headers, operation):
        # Log response status from custom header
        status_elem = envelope.find('.//{http://example.com}Status')
        if status_elem is not None:
            print(f"Operation status: {status_elem.text}")
        
        return envelope, http_headers

modifier_plugin = RequestModifierPlugin()
client = Client('http://example.com/service.wsdl', plugins=[modifier_plugin])

Multiple Plugins

from zeep import Client
from zeep.plugins import Plugin, HistoryPlugin

class MetricsPlugin(Plugin):
    def __init__(self):
        self.request_count = 0
        self.response_count = 0
    
    def egress(self, envelope, http_headers, operation, binding_options):
        self.request_count += 1
        print(f"Request #{self.request_count} to {operation.name}")
        return envelope, http_headers
    
    def ingress(self, envelope, http_headers, operation):
        self.response_count += 1
        print(f"Response #{self.response_count} from {operation.name}")
        return envelope, http_headers

# Use multiple plugins together
history = HistoryPlugin(maxlen=5)
metrics = MetricsPlugin()
auth = CustomAuthPlugin('api-key')

client = Client(
    'http://example.com/service.wsdl', 
    plugins=[auth, metrics, history]
)

# All plugins will be applied in order
result = client.service.SomeOperation(param='value')

print(f"Total requests: {metrics.request_count}")
print(f"Total responses: {metrics.response_count}")

Plugin for Error Handling

from zeep import Client
from zeep.plugins import Plugin
from zeep.exceptions import Fault

class ErrorHandlingPlugin(Plugin):
    def __init__(self):
        self.error_count = 0
    
    def ingress(self, envelope, http_headers, operation):
        # Check for SOAP faults and custom error handling
        fault = envelope.find('.//{http://schemas.xmlsoap.org/soap/envelope/}Fault')
        if fault is not None:
            self.error_count += 1
            print(f"SOAP fault detected in {operation.name}")
            
            # Could log to external system, send alerts, etc.
            
        return envelope, http_headers

error_plugin = ErrorHandlingPlugin()
client = Client('http://example.com/service.wsdl', plugins=[error_plugin])

try:
    result = client.service.OperationThatMightFail(param='value')
except Fault as e:
    print(f"Total errors encountered: {error_plugin.error_count}")
    raise

Plugin Context and State

from zeep import Client
from zeep.plugins import Plugin
import time

class TimingPlugin(Plugin):
    def __init__(self):
        self.operation_times = {}
        self._start_times = {}
    
    def egress(self, envelope, http_headers, operation, binding_options):
        # Record start time for this operation
        self._start_times[id(envelope)] = time.time()
        return envelope, http_headers
    
    def ingress(self, envelope, http_headers, operation):
        # Calculate operation duration
        envelope_id = id(envelope)
        if envelope_id in self._start_times:
            duration = time.time() - self._start_times.pop(envelope_id)
            
            op_name = operation.name
            if op_name not in self.operation_times:
                self.operation_times[op_name] = []
            
            self.operation_times[op_name].append(duration)
            print(f"{op_name} took {duration:.2f} seconds")
        
        return envelope, http_headers
    
    def get_average_time(self, operation_name):
        times = self.operation_times.get(operation_name, [])
        return sum(times) / len(times) if times else 0

timing_plugin = TimingPlugin()
client = Client('http://example.com/service.wsdl', plugins=[timing_plugin])

# Make several calls
for i in range(3):
    result = client.service.SomeOperation(param=f'value{i}')

# Check performance metrics
avg_time = timing_plugin.get_average_time('SomeOperation')
print(f"Average operation time: {avg_time:.2f} seconds")

Install with Tessl CLI