forked from amcintosh/freshbooks-python-sdk
/
resource.py
92 lines (77 loc) · 3.16 KB
/
resource.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
import json
from types import SimpleNamespace
from typing import Any, Dict, Optional
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
class HttpVerbs(object):
GET = "GET"
POST = "POST"
PUT = "PUT"
PATCH = "PATCH"
DELETE = "DELETE"
HEAD = "HEAD"
class Resource:
API_RETRIES = 3
"""Default number of retries"""
def __init__(self, client_config: SimpleNamespace):
self.base_url = client_config.base_url
self.access_token = client_config.access_token
self.user_agent = client_config.user_agent
self.timeout = client_config.timeout
self.session = self._config_session(client_config.auto_retry)
def _config_session(self, auto_retry: bool) -> requests.Session:
session = requests.Session()
if auto_retry:
retry = Retry( # type: ignore
total=self.API_RETRIES,
backoff_factor=0.3,
allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"],
status_forcelist=[400, 408, 429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
def headers(self, method: str) -> Dict[str, str]:
"""Get headers required for API calls"""
headers = {
"Authorization": f"Bearer {self.access_token}",
"user-agent": self.user_agent
}
if method in [HttpVerbs.POST, HttpVerbs.PUT, HttpVerbs.PATCH]:
headers["Content-Type"] = "application/json"
return headers
def _send_request(self, uri: str, method: str, data: Optional[dict] = None) -> Any:
payload = None
if method is HttpVerbs.GET:
session = self.session.get
elif method is HttpVerbs.POST:
session = self.session.post # type: ignore
payload = json.dumps(data)
elif method is HttpVerbs.PUT:
session = self.session.put # type: ignore
payload = json.dumps(data)
elif method is HttpVerbs.PATCH: # pragma: no cover
session = self.session.patch # type: ignore
payload = json.dumps(data)
elif method is HttpVerbs.DELETE:
session = self.session.delete
elif method is HttpVerbs.HEAD: # pragma: no cover
session = self.session.head
try:
res = session(uri, data=payload, headers=self.headers(method), timeout=self.timeout)
except requests.exceptions.RetryError:
adapter = HTTPAdapter()
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
res = session(uri, data=payload, headers=self.headers(method), timeout=self.timeout)
return res
def _build_query_string(self, builders: Any) -> str:
query_string = ""
if builders:
for builder in builders:
query_string += builder.build(self.__class__.__name__)
if query_string:
query_string = "?" + query_string[1:]
return query_string