Repository navigation
Expand file tree
/
Copy pathrequest.py
More file actions
172 lines (146 loc) · 6.52 KB
/
Copy pathrequest.py
File metadata and controls
172 lines (146 loc) · 6.52 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
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
# -*- coding: utf-8 -*-
import json
import logging
import os
from datetime import datetime
import certifi
import requests
from pytz import utc
from . import errors
logger = logging.getLogger('intercom.request')
def configure_timeout():
"""Configure the request timeout."""
timeout = os.getenv('INTERCOM_REQUEST_TIMEOUT', '90')
try:
return int(timeout)
except ValueError:
logger.warning('%s is not a valid timeout value.', timeout)
return 90
class Request(object):
timeout = configure_timeout()
def __init__(self, http_method, path, http_session=None):
self.http_method = http_method
self.path = path
self.http_session = http_session
def execute(self, base_url, auth, params):
return self.send_request_to_path(base_url, auth, params)
def send_request_to_path(self, base_url, auth, params=None):
""" Construct an API request, send it to the API, and parse the
response. """
from intercom import __version__
req_params = {}
# full URL
url = base_url + self.path
headers = {
'User-Agent': 'python-intercom/' + __version__,
'AcceptEncoding': 'gzip, deflate',
'Accept': 'application/json'
}
if self.http_method in ('POST', 'PUT', 'DELETE'):
headers['content-type'] = 'application/json'
req_params['data'] = json.dumps(params, cls=ResourceEncoder)
elif self.http_method == 'GET':
req_params['params'] = params
req_params['headers'] = headers
# request logging
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Sending %s request to: %s", self.http_method, url)
logger.debug(" headers: %s", headers)
if self.http_method == 'GET':
logger.debug(" params: %s", req_params['params'])
else:
logger.debug(" params: %s", req_params['data'])
if self.http_session is None:
resp = requests.request(
self.http_method, url, timeout=self.timeout,
auth=auth, verify=certifi.where(), **req_params)
else:
resp = self.http_session.request(
self.http_method, url, timeout=self.timeout,
auth=auth, verify=certifi.where(), **req_params)
# response logging
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response received from %s", url)
logger.debug(" encoding=%s status:%s",
resp.encoding, resp.status_code)
logger.debug(" content:\n%s", resp.content)
parsed_body = self.parse_body(resp)
self.raise_errors_on_failure(resp)
self.set_rate_limit_details(resp)
return parsed_body
def parse_body(self, resp):
if resp.content and resp.content.strip():
try:
# use supplied or inferred encoding to decode the
# response content
decoded_body = resp.content.decode(
resp.encoding or resp.apparent_encoding)
body = json.loads(decoded_body)
if body.get('type') == 'error.list':
self.raise_application_errors_on_failure(body, resp.status_code) # noqa
return body
except ValueError:
self.raise_errors_on_failure(resp)
def set_rate_limit_details(self, resp):
rate_limit_details = {}
headers = resp.headers
limit = headers.get('x-ratelimit-limit', None)
remaining = headers.get('x-ratelimit-remaining', None)
reset = headers.get('x-ratelimit-reset', None)
if limit:
rate_limit_details['limit'] = int(limit)
if remaining:
rate_limit_details['remaining'] = int(remaining)
if reset:
reset_at = datetime.utcfromtimestamp(int(reset)).replace(tzinfo=utc)
rate_limit_details['reset_at'] = reset_at
self.rate_limit_details = rate_limit_details
def raise_errors_on_failure(self, resp):
if resp.status_code == 404:
raise errors.ResourceNotFound('Resource Not Found')
elif resp.status_code == 401:
raise errors.AuthenticationError('Unauthorized')
elif resp.status_code == 403:
raise errors.AuthenticationError('Forbidden')
elif resp.status_code == 500:
raise errors.ServerError('Server Error')
elif resp.status_code == 502:
raise errors.BadGatewayError('Bad Gateway Error')
elif resp.status_code == 503:
raise errors.ServiceUnavailableError('Service Unavailable')
def raise_application_errors_on_failure(self, error_list_details, http_code): # noqa
# Currently, we don't support multiple errors
error_details = error_list_details['errors'][0]
error_code = error_details.get('type')
if error_code is None:
error_code = error_details.get('code')
error_context = {
'http_code': http_code,
'application_error_code': error_code
}
error_class = errors.error_codes.get(error_code)
if error_class is None:
# unexpected error
if error_code:
message = self.message_for_unexpected_error_with_type(
error_details, http_code)
else:
message = self.message_for_unexpected_error_without_type(
error_details, http_code)
error_class = errors.UnexpectedError
else:
message = error_details.get('message')
raise error_class(message, error_context)
def message_for_unexpected_error_with_type(self, error_details, http_code): # noqa
error_type = error_details.get('type')
message = error_details.get('message')
return "The error of type '%s' is not recognized. It occurred with the message: %s and http_code: '%s'. Please contact Intercom with these details." % (error_type, message, http_code) # noqa
def message_for_unexpected_error_without_type(self, error_details, http_code): # noqa
message = error_details['message']
return "An unexpected error occured. It occurred with the message: %s and http_code: '%s'. Please contact Intercom with these details." % (message, http_code) # noqa
class ResourceEncoder(json.JSONEncoder):
def default(self, o):
if hasattr(o, 'attributes'):
# handle API resources
return o.attributes
return super(ResourceEncoder, self).default(o)