0
# Transport and Settings
1
2
Configuration for HTTP transport, caching, session management, and XML parsing settings. Controls how zeep communicates with SOAP services and processes XML documents.
3
4
## Capabilities
5
6
### Transport Class
7
8
HTTP transport handler for SOAP requests with session management and caching support.
9
10
```python { .api }
11
class Transport:
12
def __init__(
13
self,
14
cache=None,
15
timeout: int = 300,
16
operation_timeout: int = None,
17
session=None
18
):
19
"""
20
Create HTTP transport for SOAP communication.
21
22
Parameters:
23
- cache: Cache backend instance (InMemoryCache, SqliteCache, etc.)
24
- timeout: Timeout for loading WSDL/XSD documents (seconds)
25
- operation_timeout: Timeout for SOAP operations (seconds, None=no timeout)
26
- session: Custom requests.Session instance
27
"""
28
29
def get(self, address: str, params: dict, headers: dict):
30
"""
31
Execute HTTP GET request.
32
33
Parameters:
34
- address: Target URL
35
- params: Query parameters
36
- headers: HTTP headers
37
38
Returns:
39
requests.Response object
40
"""
41
42
def post(self, address: str, message: str, headers: dict):
43
"""
44
Execute HTTP POST request.
45
46
Parameters:
47
- address: Target URL
48
- message: Request body content
49
- headers: HTTP headers
50
51
Returns:
52
requests.Response object
53
"""
54
55
def post_xml(self, address: str, envelope, headers: dict):
56
"""
57
POST XML envelope to SOAP endpoint.
58
59
Parameters:
60
- address: SOAP endpoint URL
61
- envelope: XML envelope (lxml.etree._Element)
62
- headers: HTTP headers
63
64
Returns:
65
requests.Response object
66
"""
67
```
68
69
### Async Transport
70
71
Asynchronous HTTP transport using httpx for non-blocking SOAP operations.
72
73
```python { .api }
74
class AsyncTransport(Transport):
75
"""
76
Asynchronous HTTP transport using httpx.
77
78
Provides same interface as Transport but with async methods.
79
Requires 'httpx' package to be installed.
80
"""
81
82
async def get(self, address: str, params: dict, headers: dict):
83
"""Async HTTP GET request."""
84
85
async def post(self, address: str, message: str, headers: dict):
86
"""Async HTTP POST request."""
87
88
async def post_xml(self, address: str, envelope, headers: dict):
89
"""Async XML POST to SOAP endpoint."""
90
```
91
92
### Settings Class
93
94
Configuration object for XML parsing behavior and security settings.
95
96
```python { .api }
97
class Settings:
98
def __init__(
99
self,
100
strict: bool = True,
101
raw_response: bool = False,
102
force_https: bool = True,
103
extra_http_headers: list = None,
104
xml_huge_tree: bool = False,
105
forbid_dtd: bool = False,
106
forbid_entities: bool = True,
107
forbid_external: bool = True,
108
xsd_ignore_sequence_order: bool = False
109
):
110
"""
111
Create settings for XML parsing and transport behavior.
112
113
Parameters:
114
- strict: Enable strict XML parsing mode
115
- raw_response: Return raw response without XML parsing
116
- force_https: Force HTTPS for all connections when WSDL is HTTPS
117
- extra_http_headers: Additional HTTP headers for all requests
118
- xml_huge_tree: Enable support for very large XML documents
119
- forbid_dtd: Forbid DTD processing for security
120
- forbid_entities: Forbid XML entities for security
121
- forbid_external: Forbid external resource access for security
122
- xsd_ignore_sequence_order: Ignore XSD sequence ordering (workaround)
123
"""
124
125
def __call__(self, **options):
126
"""
127
Context manager for temporary settings override.
128
129
Parameters:
130
- **options: Settings to temporarily override
131
132
Returns:
133
Context manager for scoped settings
134
"""
135
```
136
137
### Cache Backends
138
139
Various caching implementations for improving SOAP performance.
140
141
```python { .api }
142
class Base:
143
"""Base cache backend interface."""
144
145
def add(self, url: str, content: bytes):
146
"""Add content to cache for given URL."""
147
148
def get(self, url: str):
149
"""Retrieve cached content for URL."""
150
151
class InMemoryCache(Base):
152
"""In-memory cache backend."""
153
154
def __init__(self, timeout: int = 3600):
155
"""
156
Parameters:
157
- timeout: Cache expiration time in seconds
158
"""
159
160
class SqliteCache(Base):
161
"""SQLite-based persistent cache backend."""
162
163
def __init__(self, path: str = None, timeout: int = 3600):
164
"""
165
Parameters:
166
- path: SQLite database file path
167
- timeout: Cache expiration time in seconds
168
"""
169
```
170
171
## Usage Examples
172
173
### Custom Transport Configuration
174
175
```python
176
from zeep import Client, Transport
177
from zeep.cache import InMemoryCache
178
import requests
179
180
# Custom session with authentication
181
session = requests.Session()
182
session.auth = ('username', 'password')
183
session.headers.update({'User-Agent': 'MyApp/1.0'})
184
185
# Transport with caching and custom session
186
transport = Transport(
187
cache=InMemoryCache(timeout=3600),
188
timeout=30,
189
operation_timeout=120,
190
session=session
191
)
192
193
client = Client('http://example.com/service.wsdl', transport=transport)
194
```
195
196
### Settings Configuration
197
198
```python
199
from zeep import Client, Settings
200
201
# Security-focused settings
202
settings = Settings(
203
strict=True,
204
forbid_dtd=True,
205
forbid_entities=True,
206
forbid_external=True,
207
force_https=True,
208
extra_http_headers=[
209
('X-API-Key', 'your-api-key'),
210
('Accept-Encoding', 'gzip, deflate')
211
]
212
)
213
214
client = Client('https://secure-service.com/service.wsdl', settings=settings)
215
```
216
217
### Temporary Settings Override
218
219
```python
220
from zeep import Client, Settings
221
222
client = Client('http://example.com/service.wsdl')
223
224
# Temporarily disable strict parsing for problematic service
225
with client.settings(strict=False, raw_response=True):
226
raw_result = client.service.ProblematicOperation(param='value')
227
# Process raw XML response manually
228
229
# Settings return to normal after context
230
normal_result = client.service.NormalOperation(param='value')
231
```
232
233
### Persistent Caching
234
235
```python
236
from zeep import Client, Transport
237
from zeep.cache import SqliteCache
238
import os
239
240
# Use SQLite cache in user cache directory
241
cache_dir = os.path.expanduser('~/.cache/zeep')
242
os.makedirs(cache_dir, exist_ok=True)
243
244
transport = Transport(
245
cache=SqliteCache(
246
path=os.path.join(cache_dir, 'soap_cache.db'),
247
timeout=86400 # 24 hours
248
)
249
)
250
251
client = Client('http://example.com/service.wsdl', transport=transport)
252
```
253
254
### Async Transport Usage
255
256
```python
257
import asyncio
258
from zeep import AsyncClient
259
from zeep.transports import AsyncTransport
260
from zeep.cache import InMemoryCache
261
262
async def async_soap_call():
263
transport = AsyncTransport(
264
cache=InMemoryCache(),
265
timeout=30,
266
operation_timeout=60
267
)
268
269
async with AsyncClient('http://example.com/service.wsdl', transport=transport) as client:
270
result = await client.service.AsyncOperation(param='value')
271
return result
272
273
result = asyncio.run(async_soap_call())
274
```