-
Notifications
You must be signed in to change notification settings - Fork 133
/
util.py
276 lines (228 loc) · 8.68 KB
/
util.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
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
"""Utility classes and functions"""
from __future__ import absolute_import, unicode_literals
from urllib.parse import quote
import contextlib
import json
import requests
import time
import jwt
API_VERSION_1 = 1
API_VERSION_2 = 2
class ApiClient(object):
"""Simple wrapper for REST API requests"""
def __init__(self, base_uri=None, timeout=15, **kwargs):
"""Setup a new API Client
:param base_uri: The base URI to the API
:param timeout: The timeout to use for requests
:param kwargs: Any other attributes. These will be added as
attributes to the ApiClient object.
"""
self.base_uri = base_uri
self.timeout = timeout
for k, v in kwargs.items():
setattr(self, k, v)
@property
def timeout(self):
"""The timeout"""
return self._timeout
@timeout.setter
def timeout(self, value):
"""The default timeout"""
if value is not None:
try:
value = int(value)
except ValueError:
raise ValueError("timeout value must be an integer")
self._timeout = value
@property
def base_uri(self):
"""The base_uri"""
return self._base_uri
@base_uri.setter
def base_uri(self, value):
"""The default base_uri"""
if value and value.endswith("/"):
value = value[:-1]
self._base_uri = value
def url_for(self, endpoint):
"""Get the URL for the given endpoint
:param endpoint: The endpoint
:return: The full URL for the endpoint
"""
if not endpoint.startswith("/"):
endpoint = "/{}".format(endpoint)
if endpoint.endswith("/"):
endpoint = endpoint[:-1]
return self.base_uri + endpoint
def get_request(self, endpoint, params=None, headers=None):
"""Helper function for GET requests
:param endpoint: The endpoint
:param params: The URL parameters
:param headers: request headers
:return: The :class:``requests.Response`` object for this request
"""
if headers is None and self.config.get("version") == API_VERSION_2:
headers = {"Authorization": "Bearer {}".format(self.config.get("token"))}
return requests.get(
self.url_for(endpoint), params=params, headers=headers, timeout=self.timeout
)
def post_request(
self, endpoint, params=None, data=None, headers=None, cookies=None
):
"""Helper function for POST requests
:param endpoint: The endpoint
:param params: The URL parameters
:param data: The data (either as a dict or dumped JSON string) to
include with the POST
:param headers: request headers
:param cookies: request cookies
:return: The :class:``requests.Response`` object for this request
"""
if data and not is_str_type(data):
data = json.dumps(data)
if headers is None and self.config.get("version") == API_VERSION_2:
headers = {
"Authorization": "Bearer {}".format(self.config.get("token")),
"Content-Type": "application/json",
}
return requests.post(
self.url_for(endpoint),
params=params,
data=data,
headers=headers,
cookies=cookies,
timeout=self.timeout,
)
def patch_request(
self, endpoint, params=None, data=None, headers=None, cookies=None
):
"""Helper function for PATCH requests
:param endpoint: The endpoint
:param params: The URL parameters
:param data: The data (either as a dict or dumped JSON string) to
include with the PATCH
:param headers: request headers
:param cookies: request cookies
:return: The :class:``requests.Response`` object for this request
"""
if data and not is_str_type(data):
data = json.dumps(data)
if headers is None and self.config.get("version") == API_VERSION_2:
headers = {
"Authorization": "Bearer {}".format(self.config.get("token")),
"Content-Type": "application/json",
}
return requests.patch(
self.url_for(endpoint),
params=params,
data=data,
headers=headers,
cookies=cookies,
timeout=self.timeout,
)
def delete_request(
self, endpoint, params=None, data=None, headers=None, cookies=None
):
"""Helper function for DELETE requests
:param endpoint: The endpoint
:param params: The URL parameters
:param data: The data (either as a dict or dumped JSON string) to
include with the DELETE
:param headers: request headers
:param cookies: request cookies
:return: The :class:``requests.Response`` object for this request
"""
if data and not is_str_type(data):
data = json.dumps(data)
if headers is None and self.config.get("version") == API_VERSION_2:
headers = {"Authorization": "Bearer {}".format(self.config.get("token"))}
return requests.delete(
self.url_for(endpoint),
params=params,
data=data,
headers=headers,
cookies=cookies,
timeout=self.timeout,
)
def put_request(self, endpoint, params=None, data=None, headers=None, cookies=None):
"""Helper function for PUT requests
:param endpoint: The endpoint
:param params: The URL paramaters
:param data: The data (either as a dict or dumped JSON string) to
include with the PUT
:param headers: request headers
:param cookies: request cookies
:return: The :class:``requests.Response`` object for this request
"""
if data and not is_str_type(data):
data = json.dumps(data)
if headers is None and self.config.get("version") == API_VERSION_2:
headers = {"Authorization": "Bearer {}".format(self.config.get("token"))}
return requests.put(
self.url_for(endpoint),
params=params,
data=data,
headers=headers,
cookies=cookies,
timeout=self.timeout,
)
@contextlib.contextmanager
def ignored(*exceptions):
"""Simple context manager to ignore expected Exceptions
:param \*exceptions: The exceptions to safely ignore
"""
try:
yield
except exceptions:
pass
def is_str_type(val):
"""Check whether the input is of a string type.
We use this method to ensure python 2-3 capatibility.
:param val: The value to check wither it is a string
:return: In python2 it will return ``True`` if :attr:`val` is either an
instance of str or unicode. In python3 it will return ``True`` if
it is an instance of str
"""
with ignored(NameError):
return isinstance(val, basestring)
return isinstance(val, str)
def require_keys(d, keys, allow_none=True):
"""Require that the object have the given keys
:param d: The dict the check
:param keys: The keys to check :attr:`obj` for. This can either be a single
string, or an iterable of strings
:param allow_none: Whether ``None`` values are allowed
:raises:
:ValueError: If any of the keys are missing from the obj
"""
if is_str_type(keys):
keys = [keys]
for k in keys:
if k not in d:
raise ValueError("'{}' must be set".format(k))
if not allow_none and d[k] is None:
raise ValueError("'{}' cannot be None".format(k))
return True
def date_to_str(d):
"""Convert date and datetime objects to a string
Note, this does not do any timezone conversion.
:param d: The :class:`datetime.date` or :class:`datetime.datetime` to
convert to a string
:returns: The string representation of the date
"""
return d.strftime("%Y-%m-%dT%H:%M:%SZ")
def generate_jwt(key, secret):
header = {"alg": "HS256", "typ": "JWT"}
payload = {"iss": key, "exp": int(time.time() + 3600)}
token = jwt.encode(payload, secret, algorithm="HS256", headers=header)
return token.decode("utf-8")
def encode_uuid(val):
"""Encode UUID as described by ZOOM API documentation
> Note: Please double encode your UUID when using this API if the UUID
> begins with a '/'or contains ‘//’ in it.
:param val: The UUID to encode
:returns: The encoded UUID
"""
if val[0] == "/" or "//" in val:
val = quote(quote(val, safe=""), safe="")
return val