Repository navigation
Expand file tree
/
Copy pathclient.py
More file actions
82 lines (63 loc) · 2.35 KB
/
Copy pathclient.py
File metadata and controls
82 lines (63 loc) · 2.35 KB
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
import time
from urllib.parse import urljoin
import requests
from requests.exceptions import RequestException
import inkit
from inkit.exceptions import InkitClientException, InkitResponseException
from inkit.response_object import ResponseObject
HOST = 'https://api.inkit.com/v1'
USER_AGENT = 'Inkit SDK'
TIMEOUT = 20
class Client:
def __init__(self):
self._session = self._build_session()
@staticmethod
def _build_session():
session = requests.Session()
session.headers.update({
'User-Agent': USER_AGENT,
'Content-Type': 'application/json'
})
return session
def send(self, path, http_method, params=None, data=None, retry=0, retry_interval=0, status_forcelist=None):
if not inkit.api_token:
raise InkitClientException(message='API Token is not specified')
if not isinstance(inkit.api_token, str):
raise InkitClientException(message=f'API Token must be a string, got {type(inkit.api_token)}')
request_data = {
'url': urljoin(HOST, path),
'timeout': TIMEOUT,
'headers': {'X-Inkit-API-Token': inkit.api_token}
}
if data:
request_data.update(data=data)
if params:
request_data.update(params=params)
method = getattr(self._session, http_method.lower())
return self._send_request(
method=method,
request_data=request_data,
retry=retry,
retry_interval=retry_interval,
status_forcelist=status_forcelist or []
)
def _send_request(self, method, request_data, retry, retry_interval, status_forcelist):
try:
resp = method(**request_data)
except RequestException as e:
raise InkitClientException(exc=e)
if resp.ok:
return ResponseObject(resp)
if retry and resp.status_code in status_forcelist:
time.sleep(retry_interval)
return self._send_request(
method=method,
request_data=request_data,
retry=retry-1,
retry_interval=retry_interval,
status_forcelist=status_forcelist
)
raise InkitResponseException(
message='API responded with invalid status code',
resp=ResponseObject(resp)
)