0
# Plugin System
1

2
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.
3

4
## Capabilities
5

6
### Base Plugin Class
7

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

10
```python { .api }
11
class Plugin:
12
    def ingress(self, envelope, http_headers: dict, operation):
13
        """
14
        Process incoming SOAP responses.
15
        
16
        Called when receiving responses from SOAP services.
17
        
18
        Parameters:
19
        - envelope: Response envelope (lxml.etree._Element)
20
        - http_headers: HTTP response headers dict
21
        - operation: Operation instance that generated the request
22
        
23
        Returns:
24
        tuple: (modified_envelope, modified_headers) or None to keep unchanged
25
        """
26

27
    def egress(self, envelope, http_headers: dict, operation, binding_options: dict):
28
        """
29
        Process outgoing SOAP requests.
30
        
31
        Called before sending requests to SOAP services.
32
        
33
        Parameters:
34
        - envelope: Request envelope (lxml.etree._Element)  
35
        - http_headers: HTTP request headers dict
36
        - operation: Operation instance being called
37
        - binding_options: Binding-specific options dict
38
        
39
        Returns:
40
        tuple: (modified_envelope, modified_headers) or None to keep unchanged
41
        """
42
```
43

44
### History Plugin
45

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

48
```python { .api }
49
class HistoryPlugin(Plugin):
50
    def __init__(self, maxlen: int = 1):
51
        """
52
        Create history tracking plugin.
53
        
54
        Parameters:
55
        - maxlen: Maximum number of request/response pairs to store
56
        """
57

58
    @property
59
    def last_sent(self):
60
        """Last sent request envelope."""
61

62
    @property  
63
    def last_received(self):
64
        """Last received response envelope."""
65
```
66

67
### Plugin Application Functions
68

69
Utility functions for applying plugins to requests and responses.
70

71
```python { .api }
72
def apply_egress(client, envelope, http_headers: dict, operation, binding_options: dict):
73
    """
74
    Apply all egress plugins to outgoing request.
75
    
76
    Parameters:
77
    - client: Client instance with plugins
78
    - envelope: Request envelope
79
    - http_headers: HTTP headers
80
    - operation: Operation being called
81
    - binding_options: Binding options
82
    
83
    Returns:
84
    tuple: (processed_envelope, processed_headers)
85
    """
86

87
def apply_ingress(client, envelope, http_headers: dict, operation):
88
    """
89
    Apply all ingress plugins to incoming response.
90
    
91
    Parameters:
92
    - client: Client instance with plugins
93
    - envelope: Response envelope  
94
    - http_headers: HTTP headers
95
    - operation: Operation that generated request
96
    
97
    Returns:
98
    tuple: (processed_envelope, processed_headers)
99
    """
100
```
101

102
## Usage Examples
103

104
### Basic Plugin Usage
105

106
```python
107
from zeep import Client
108
from zeep.plugins import HistoryPlugin
109

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

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

117
# Inspect request/response history
118
print("Last sent request:")
119
print(history.last_sent)
120

121
print("Last received response:")  
122
print(history.last_received)
123
```
124

125
### Custom Logging Plugin
126

127
```python
128
import logging
129
from zeep import Client
130
from zeep.plugins import Plugin
131

132
class LoggingPlugin(Plugin):
133
    def __init__(self, logger_name='zeep.plugin'):
134
        self.logger = logging.getLogger(logger_name)
135
    
136
    def egress(self, envelope, http_headers, operation, binding_options):
137
        self.logger.info(f"Sending request to operation: {operation.name}")
138
        self.logger.debug(f"Request envelope: {envelope}")
139
        return envelope, http_headers
140
    
141
    def ingress(self, envelope, http_headers, operation):
142
        self.logger.info(f"Received response from operation: {operation.name}")
143
        self.logger.debug(f"Response envelope: {envelope}")
144
        return envelope, http_headers
145

146
# Set up logging
147
logging.basicConfig(level=logging.INFO)
148

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

153
result = client.service.SomeOperation(param='value')
154
```
155

156
### Authentication Plugin
157

158
```python
159
from zeep import Client
160
from zeep.plugins import Plugin
161
import base64
162

163
class CustomAuthPlugin(Plugin):
164
    def __init__(self, api_key):
165
        self.api_key = api_key
166
    
167
    def egress(self, envelope, http_headers, operation, binding_options):
168
        # Add custom authentication header
169
        http_headers['X-API-Key'] = self.api_key
170
        http_headers['Authorization'] = f'Bearer {self.api_key}'
171
        return envelope, http_headers
172

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

177
result = client.service.AuthenticatedOperation(param='value')
178
```
179

180
### Request/Response Modification Plugin
181

182
```python
183
from zeep import Client
184
from zeep.plugins import Plugin
185
from lxml import etree
186

187
class RequestModifierPlugin(Plugin):
188
    def egress(self, envelope, http_headers, operation, binding_options):
189
        # Add custom SOAP header
190
        header = envelope.find('.//{http://schemas.xmlsoap.org/soap/envelope/}Header')
191
        if header is None:
192
            header = etree.Element('{http://schemas.xmlsoap.org/soap/envelope/}Header')
193
            envelope.insert(0, header)
194
        
195
        # Add custom header element
196
        custom_header = etree.SubElement(header, '{http://example.com}CustomHeader')
197
        custom_header.text = 'Custom Value'
198
        
199
        return envelope, http_headers
200
    
201
    def ingress(self, envelope, http_headers, operation):
202
        # Log response status from custom header
203
        status_elem = envelope.find('.//{http://example.com}Status')
204
        if status_elem is not None:
205
            print(f"Operation status: {status_elem.text}")
206
        
207
        return envelope, http_headers
208

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

213
### Multiple Plugins
214

215
```python
216
from zeep import Client
217
from zeep.plugins import Plugin, HistoryPlugin
218

219
class MetricsPlugin(Plugin):
220
    def __init__(self):
221
        self.request_count = 0
222
        self.response_count = 0
223
    
224
    def egress(self, envelope, http_headers, operation, binding_options):
225
        self.request_count += 1
226
        print(f"Request #{self.request_count} to {operation.name}")
227
        return envelope, http_headers
228
    
229
    def ingress(self, envelope, http_headers, operation):
230
        self.response_count += 1
231
        print(f"Response #{self.response_count} from {operation.name}")
232
        return envelope, http_headers
233

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

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

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

247
print(f"Total requests: {metrics.request_count}")
248
print(f"Total responses: {metrics.response_count}")
249
```
250

251
### Plugin for Error Handling
252

253
```python
254
from zeep import Client
255
from zeep.plugins import Plugin
256
from zeep.exceptions import Fault
257

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

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

276
try:
277
    result = client.service.OperationThatMightFail(param='value')
278
except Fault as e:
279
    print(f"Total errors encountered: {error_plugin.error_count}")
280
    raise
281
```
282

283
### Plugin Context and State
284

285
```python
286
from zeep import Client
287
from zeep.plugins import Plugin
288
import time
289

290
class TimingPlugin(Plugin):
291
    def __init__(self):
292
        self.operation_times = {}
293
        self._start_times = {}
294
    
295
    def egress(self, envelope, http_headers, operation, binding_options):
296
        # Record start time for this operation
297
        self._start_times[id(envelope)] = time.time()
298
        return envelope, http_headers
299
    
300
    def ingress(self, envelope, http_headers, operation):
301
        # Calculate operation duration
302
        envelope_id = id(envelope)
303
        if envelope_id in self._start_times:
304
            duration = time.time() - self._start_times.pop(envelope_id)
305
            
306
            op_name = operation.name
307
            if op_name not in self.operation_times:
308
                self.operation_times[op_name] = []
309
            
310
            self.operation_times[op_name].append(duration)
311
            print(f"{op_name} took {duration:.2f} seconds")
312
        
313
        return envelope, http_headers
314
    
315
    def get_average_time(self, operation_name):
316
        times = self.operation_times.get(operation_name, [])
317
        return sum(times) / len(times) if times else 0
318

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

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

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