-
-
Notifications
You must be signed in to change notification settings - Fork 583
/
transports.py
256 lines (203 loc) · 7.79 KB
/
transports.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
import logging
import os
from contextlib import contextmanager
from urllib.parse import urlparse
import requests
from requests import Response
from requests_file import FileAdapter
from zeep.exceptions import TransportError
from zeep.utils import get_media_type, get_version
from zeep.wsdl.utils import etree_to_string
try:
import httpx
except ImportError:
httpx = None
__all__ = ["AsyncTransport", "Transport"]
class Transport:
"""The transport object handles all communication to the SOAP server.
:param cache: The cache object to be used to cache GET requests
:param timeout: The timeout for loading wsdl and xsd documents.
:param operation_timeout: The timeout for operations (POST/GET). By
default this is None (no timeout).
:param session: A :py:class:`request.Session()` object (optional)
"""
def __init__(self, cache=None, timeout=300, operation_timeout=None, session=None):
self.cache = cache
self.load_timeout = timeout
self.operation_timeout = operation_timeout
self.logger = logging.getLogger(__name__)
self._close_session = not session
self.session = session or requests.Session()
self.session.mount("file://", FileAdapter())
self.session.headers["User-Agent"] = "Zeep/%s (www.python-zeep.org)" % (
get_version()
)
def get(self, address, params, headers):
"""Proxy to requests.get()
:param address: The URL for the request
:param params: The query parameters
:param headers: a dictionary with the HTTP headers.
"""
response = self.session.get(
address, params=params, headers=headers, timeout=self.operation_timeout
)
return response
def post(self, address, message, headers):
"""Proxy to requests.posts()
:param address: The URL for the request
:param message: The content for the body
:param headers: a dictionary with the HTTP headers.
"""
if self.logger.isEnabledFor(logging.DEBUG):
log_message = message
if isinstance(log_message, bytes):
log_message = log_message.decode("utf-8")
self.logger.debug("HTTP Post to %s:\n%s", address, log_message)
response = self.session.post(
address, data=message, headers=headers, timeout=self.operation_timeout
)
if self.logger.isEnabledFor(logging.DEBUG):
media_type = get_media_type(
response.headers.get("Content-Type", "text/xml")
)
if media_type == "multipart/related":
log_message = response.content
else:
log_message = response.content
if isinstance(log_message, bytes):
log_message = log_message.decode(response.encoding or "utf-8")
self.logger.debug(
"HTTP Response from %s (status: %d):\n%s",
address,
response.status_code,
log_message,
)
return response
def post_xml(self, address, envelope, headers):
"""Post the envelope xml element to the given address with the headers.
This method is intended to be overriden if you want to customize the
serialization of the xml element. By default the body is formatted
and encoded as utf-8. See ``zeep.wsdl.utils.etree_to_string``.
"""
message = etree_to_string(envelope)
return self.post(address, message, headers)
def load(self, url):
"""Load the content from the given URL"""
if not url:
raise ValueError("No url given to load")
scheme = urlparse(url).scheme
if scheme in ("http", "https", "file"):
if self.cache:
response = self.cache.get(url)
if response:
return bytes(response)
content = self._load_remote_data(url)
if self.cache:
self.cache.add(url, content)
return content
else:
with open(os.path.expanduser(url), "rb") as fh:
return fh.read()
def _load_remote_data(self, url):
self.logger.debug("Loading remote data from: %s", url)
response = self.session.get(url, timeout=self.load_timeout)
response.raise_for_status()
return response.content
@contextmanager
def settings(self, timeout=None):
"""Context manager to temporarily overrule options.
Example::
transport = zeep.Transport()
with transport.settings(timeout=10):
client.service.fast_call()
:param timeout: Set the timeout for POST/GET operations (not used for
loading external WSDL or XSD documents)
"""
old_timeout = self.operation_timeout
self.operation_timeout = timeout
yield
self.operation_timeout = old_timeout
def __del__(self):
if self._close_session:
self.session.close()
class AsyncTransport(Transport):
"""Asynchronous Transport class using httpx.
Note that loading the wsdl is still a sync process since and only the
operations can be called via async.
"""
def __init__(
self,
client=None,
wsdl_client=None,
cache=None,
timeout=300,
operation_timeout=None,
verify_ssl=True,
proxy=None,
):
if httpx is None:
raise RuntimeError("The AsyncTransport is based on the httpx module")
self._close_session = False
self.cache = cache
self.wsdl_client = wsdl_client or httpx.Client(
verify=verify_ssl,
proxies=proxy,
timeout=timeout,
)
self.client = client or httpx.AsyncClient(
verify=verify_ssl,
proxies=proxy,
timeout=operation_timeout,
)
self.logger = logging.getLogger(__name__)
self.wsdl_client.headers = {
"User-Agent": "Zeep/%s (www.python-zeep.org)" % (get_version())
}
self.client.headers = {
"User-Agent": "Zeep/%s (www.python-zeep.org)" % (get_version())
}
async def aclose(self):
await self.client.aclose()
def _load_remote_data(self, url):
response = self.wsdl_client.get(url)
result = response.read()
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
raise TransportError(status_code=response.status_code)
return result
async def post(self, address, message, headers):
self.logger.debug("HTTP Post to %s:\n%s", address, message)
response = await self.client.post(
address,
content=message,
headers=headers,
)
self.logger.debug(
"HTTP Response from %s (status: %d):\n%s",
address,
response.status_code,
response.read(),
)
return response
async def post_xml(self, address, envelope, headers):
message = etree_to_string(envelope)
response = await self.post(address, message, headers)
return self.new_response(response)
async def get(self, address, params, headers):
response = await self.client.get(
address,
params=params,
headers=headers,
)
return self.new_response(response)
def new_response(self, response):
"""Convert an aiohttp.Response object to a requests.Response object"""
body = response.read()
new = Response()
new._content = body
new.status_code = response.status_code
new.headers = response.headers
new.cookies = response.cookies
new.encoding = response.encoding
return new