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
# Licensed to Elasticsearch B.V under one or more agreements.
# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
# See the LICENSE file in the project root for more information
import time
import warnings
try:
import requests
REQUESTS_AVAILABLE = True
except ImportError:
REQUESTS_AVAILABLE = False
from .base import Connection
from ..exceptions import (
ConnectionError,
ImproperlyConfigured,
ConnectionTimeout,
SSLError,
)
from ..compat import urlencode, string_types
class RequestsHttpConnection(Connection):
"""
Connection using the `requests` library.
:arg http_auth: optional http auth information as either ':' separated
string or a tuple. Any value will be passed into requests as `auth`.
:arg use_ssl: use ssl for the connection if `True`
:arg verify_certs: whether to verify SSL certificates
:arg ssl_show_warn: show warning when verify certs is disabled
:arg ca_certs: optional path to CA bundle. By default standard requests'
bundle will be used.
:arg client_cert: path to the file containing the private key and the
certificate, or cert only if using client_key
:arg client_key: path to the file containing the private key if using
separate cert and key files (client_cert will contain only the cert)
:arg headers: any custom http headers to be add to requests
:arg http_compress: Use gzip compression
:arg cloud_id: The Cloud ID from ElasticCloud. Convenient way to connect to cloud instances.
Other host connection params will be ignored.
:arg api_key: optional API Key authentication as either base64 encoded string or a tuple.
:arg opaque_id: Send this value in the 'X-Opaque-Id' HTTP header
For tracing all requests made by this transport.
"""
def __init__(
self,
host="localhost",
port=None,
http_auth=None,
use_ssl=False,
verify_certs=True,
ssl_show_warn=True,
ca_certs=None,
client_cert=None,
client_key=None,
headers=None,
http_compress=None,
cloud_id=None,
api_key=None,
opaque_id=None,
**kwargs
):
if not REQUESTS_AVAILABLE:
raise ImproperlyConfigured(
"Please install requests to use RequestsHttpConnection."
)
# Initialize Session so .headers works before calling super().__init__().
self.session = requests.Session()
for key in list(self.session.headers):
self.session.headers.pop(key)
super(RequestsHttpConnection, self).__init__(
host=host,
port=port,
use_ssl=use_ssl,
headers=headers,
http_compress=http_compress,
cloud_id=cloud_id,
api_key=api_key,
opaque_id=opaque_id,
**kwargs
)
if not self.http_compress:
# Need to set this to 'None' otherwise Requests adds its own.
self.session.headers["accept-encoding"] = None
if http_auth is not None:
if isinstance(http_auth, (tuple, list)):
http_auth = tuple(http_auth)
elif isinstance(http_auth, string_types):
http_auth = tuple(http_auth.split(":", 1))
self.session.auth = http_auth
self.base_url = "%s%s" % (self.host, self.url_prefix,)
self.session.verify = verify_certs
if not client_key:
self.session.cert = client_cert
elif client_cert:
# cert is a tuple of (certfile, keyfile)
self.session.cert = (client_cert, client_key)
if ca_certs:
if not verify_certs:
raise ImproperlyConfigured(
"You cannot pass CA certificates when verify SSL is off."
)
self.session.verify = ca_certs
if not ssl_show_warn:
requests.packages.urllib3.disable_warnings()
if self.use_ssl and not verify_certs and ssl_show_warn:
warnings.warn(
"Connecting to %s using SSL with verify_certs=False is insecure."
% self.host
)
def perform_request(
self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None
):
url = self.base_url + url
headers = headers or {}
if params:
url = "%s?%s" % (url, urlencode(params or {}))
orig_body = body
if self.http_compress and body:
body = self._gzip_compress(body)
headers["content-encoding"] = "gzip"
start = time.time()
request = requests.Request(method=method, headers=headers, url=url, data=body)
prepared_request = self.session.prepare_request(request)
settings = self.session.merge_environment_settings(
prepared_request.url, {}, None, None, None
)
send_kwargs = {"timeout": timeout or self.timeout}
send_kwargs.update(settings)
try:
response = self.session.send(prepared_request, **send_kwargs)
duration = time.time() - start
raw_data = response.content.decode("utf-8", "surrogatepass")
except Exception as e:
self.log_request_fail(
method,
url,
prepared_request.path_url,
body,
time.time() - start,
exception=e,
)
if isinstance(e, requests.exceptions.SSLError):
raise SSLError("N/A", str(e), e)
if isinstance(e, requests.Timeout):
raise ConnectionTimeout("TIMEOUT", str(e), e)
raise ConnectionError("N/A", str(e), e)
# raise warnings if any from the 'Warnings' header.
warnings_headers = (
(response.headers["warning"],) if "warning" in response.headers else ()
)
self._raise_warnings(warnings_headers)
# raise errors based on http status codes, let the client handle those if needed
if (
not (200 <= response.status_code < 300)
and response.status_code not in ignore
):
self.log_request_fail(
method,
url,
response.request.path_url,
orig_body,
duration,
response.status_code,
raw_data,
)
self._raise_error(response.status_code, raw_data)
self.log_request_success(
method,
url,
response.request.path_url,
orig_body,
response.status_code,
raw_data,
duration,
)
return response.status_code, response.headers, raw_data
@property
def headers(self):
return self.session.headers
def close(self):
"""
Explicitly closes connections
"""
self.session.close()