forked from mollie/mollie-api-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
base.py
92 lines (75 loc) · 3.67 KB
/
base.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 ..error import RequestSetupError, ResponseError, ResponseHandlingError
from ..objects.list import List
class Base(object):
REST_CREATE = 'POST'
REST_UPDATE = 'PATCH'
REST_READ = 'GET'
REST_LIST = 'GET'
REST_DELETE = 'DELETE'
DEFAULT_LIMIT = 10
def __init__(self, client):
self.client = client
def get_resource_object(self, result):
raise NotImplementedError()
def get_resource_name(self):
return self.__class__.__name__.lower()
def rest_create(self, data, params=None):
path = self.get_resource_name()
result = self.perform_api_call(self.REST_CREATE, path, data, params)
return self.get_resource_object(result)
def rest_read(self, resource_id, params=None):
path = self.get_resource_name() + '/' + str(resource_id)
result = self.perform_api_call(self.REST_READ, path, params=params)
return self.get_resource_object(result)
def rest_update(self, resource_id, data, params=None):
path = self.get_resource_name() + '/' + str(resource_id)
result = self.perform_api_call(self.REST_UPDATE, path, data, params)
return self.get_resource_object(result)
def rest_delete(self, resource_id, params=None):
path = self.get_resource_name() + '/' + str(resource_id)
return self.perform_api_call(self.REST_DELETE, path, params=params)
def rest_list(self, params=None):
path = self.get_resource_name()
result = self.perform_api_call(self.REST_LIST, path, params=params)
return List(result, self.get_resource_object({}).__class__)
def create(self, data=None, **params):
if data is not None:
try:
data = json.dumps(data)
except Exception as err:
raise RequestSetupError("Error encoding parameters into JSON: '{error}'.".format(error=err))
return self.rest_create(data, params)
def get(self, resource_id, **params):
return self.rest_read(resource_id, params)
def update(self, resource_id, data=None, **params):
if data is not None:
try:
data = json.dumps(data)
except Exception as err:
raise RequestSetupError("Error encoding parameters into JSON: '{error}'.".format(error=err))
return self.rest_update(resource_id, data, params)
def delete(self, resource_id):
return self.rest_delete(resource_id)
def list(self, **params):
return self.rest_list(params)
def perform_api_call(self, http_method, path, data=None, params=None):
resp = self.client.perform_http_call(http_method, path, data, params)
# Mollie returns unicode data in responses, but doesn't set the Content-Type header correctly
resp.encoding = 'utf-8'
try:
result = resp.json() if resp.status_code != 204 else {}
except Exception:
raise ResponseHandlingError(
"Unable to decode Mollie API response (status code: {status}): '{response}'.".format(
status=resp.status_code, response=resp.text))
if resp.status_code < 200 or resp.status_code > 299:
if 'status' in result and (result['status'] < 200 or result['status'] > 299):
# the factory will return the appropriate ResponseError subclass based on the result
raise ResponseError.factory(result)
else:
raise ResponseHandlingError(
"Received HTTP error from Mollie API, but no status in payload "
"(status code: {status}): '{response}'.".format(
status=resp.status_code, response=resp.text))
return result