Repository navigation
Expand file tree
/
Copy pathrequest.py
More file actions
129 lines (112 loc) · 4.84 KB
/
Copy pathrequest.py
File metadata and controls
129 lines (112 loc) · 4.84 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
# -*- coding: utf-8 -*-
from . import errors
from datetime import datetime
import certifi
import json
import requests
class Request(object):
timeout = 10
@classmethod
def send_request_to_path(cls, method, url, auth, params=None):
""" Construct an API request, send it to the API, and parse the
response. """
from intercom import __version__
req_params = {}
headers = {
'User-Agent': 'python-intercom/' + __version__,
'Accept': 'application/json'
}
if method in ('POST', 'PUT', 'DELETE'):
headers['content-type'] = 'application/json'
req_params['data'] = json.dumps(params, cls=ResourceEncoder)
elif method == 'GET':
req_params['params'] = params
req_params['headers'] = headers
resp = requests.request(
method, url, timeout=cls.timeout,
auth=auth, verify=certifi.where(), **req_params)
cls.raise_errors_on_failure(resp)
cls.set_rate_limit_details(resp)
if resp.content:
return cls.parse_body(resp)
@classmethod
def parse_body(cls, resp):
try:
# use supplied encoding to decode the response content
decoded_body = resp.content.decode(resp.encoding)
if not decoded_body: # return early for empty responses (issue-72)
return
body = json.loads(decoded_body)
if body.get('type') == 'error.list':
cls.raise_application_errors_on_failure(body, resp.status_code)
return body
except ValueError:
cls.raise_errors_on_failure(resp)
@classmethod
def set_rate_limit_details(cls, 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:
rate_limit_details['reset_at'] = datetime.fromtimestamp(int(reset))
from intercom import Intercom
Intercom.rate_limit_details = rate_limit_details
@classmethod
def raise_errors_on_failure(cls, 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')
@classmethod
def raise_application_errors_on_failure(cls, 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 = cls.message_for_unexpected_error_with_type(
error_details, http_code)
else:
message = cls.message_for_unexpected_error_without_type(
error_details, http_code)
error_class = errors.UnexpectedError
else:
message = error_details['message']
raise error_class(message, error_context)
@classmethod
def message_for_unexpected_error_with_type(cls, error_details, http_code): # noqa
error_type = error_details['type']
message = error_details['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
@classmethod
def message_for_unexpected_error_without_type(cls, 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)