-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathsuiteCRMService.py
224 lines (193 loc) · 9.93 KB
/
suiteCRMService.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
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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
from .module import Module
import requests
import json
# SuiteCRM Web Service class to interact with the SuiteCRM v8 API
class SuiteCRMService:
PARAMETER_MUST_NOT_BE_NONE = "Parameter {0} must not be None"
PARAMETER_MUST_BE_OF_TYPE = "Parameter {0} must be of type {1}"
#Constructor for the WebService. Gets the access token with the given clientId and clientSecret
#Parameters
#----------
#host : string
# The host url of the service
#client_id : string
# The client id for the authentication
#client_secret : string
# The client secret for the authentication
def __init__(self, host, client_id, client_secret):
if host == None or host == "":
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("host"))
if client_id == None or client_id == "":
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("client_id"))
if client_secret == None or client_secret == "":
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("client_secret"))
self._host = host
self._auth_header = self._get_auth_header(client_id, client_secret)
self._auth_header['Content-type'] = 'application/vnd.api+json'
#Get the authorization token and create and authentication header
#Parameters
#----------
#client_id : string
# The client id for the authentication
#client_secret : string
# The client secret for the authentication
def _get_auth_header(self, client_id, client_secret) -> dict:
body = {"grant_type": "client_credentials","client_id": client_id,"client_secret": client_secret}
response = requests.post("{0}/Api/access_token".format(self._host), data = body)
if(response.status_code == 200):
access_token = response.json().get("access_token")
return {'Authorization': 'Bearer {0}'.format(access_token)}
else:
raise RuntimeError("Not able to get token {0}".format(response.text))
#Get all available modules
def get_modules(self) -> requests.Response:
return requests.get("{0}/Api/V8/meta/modules".format(self._host), headers=self._auth_header)
#Get fields for a certain module
#Parameters
#----------
#module : Module
# The module you want to get the data for.
def get_module_fields(self, module) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
return requests.get("{0}/Api/V8/meta/fields/{1}".format(self._host, module.value), headers=self._auth_header)
#get data for a certain module
#Parameters
#----------
#module : Module
# The module you want to get the data for.
#fields : list
# The fields to be returned. If None then all fields will be returned.
#filter : Filter
# The filter you want ot set. Default is None
#pagination: Pagination
# The pagination settings to be set. Default page size is 50
def get_data(self, module, fields=None, filter=None, pagination=None) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
seperator = ","
if module != None and type(fields) == list and len(fields) > 0:
fields = "fields[{0}]={1}".format(module.value, seperator.join(fields))
response = requests.get("{0}/Api/V8/module/{1}{2}".format(self._host, module.value, self._build_query_params(fields, filter, pagination)), headers=self._auth_header)
return response
#get relationship data for a certain module
#Parameters
#----------
#module : Module
# The module you want to get the data for.
#id: str
# The ID of the object to get the relationship for
#relationship : Module
# The relationship module to get the data for. (e.g. CONTACTS for contacts of an OPPORTUNITY)
#fields : list
# The fields to be returned. If None then all fields will be returned.
#filter : Filter
# The filter you want ot set. Default is None
#pagination: Pagination
# The pagination settings to be set. Default page size is 50
def get_relationship_data(self, module, id, relationship, fields=None, filter=None, pagination=None) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
seperator = ","
if module != None and type(fields) == list and len(fields) > 0:
fields = "fields[{0}]={1}".format(module.value, seperator.join(fields))
response = requests.get("{0}/Api/V8/module/{1}/{2}/relationships/{3}{4}".format(self._host, module.value, id, relationship.value.lower(), self._build_query_params(fields, filter, pagination)), headers=self._auth_header)
return response
#Insert a record
#Parameters
#----------
#module : Module
# The module you want to get the data for.
#attributes : dict[str, object]
# The attributes to be updated
def insert_data(self, module, attributes) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
if attributes is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("attributes"))
if not isinstance(attributes, dict):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("attributes", "dict"))
body = {'data': { 'type': module.name.lower().capitalize(), 'attributes': attributes}}
json_data = json.dumps(body, default=lambda o: o.__dict__, sort_keys=False)
return requests.post('{0}/Api/V8/module'.format(self._host), data = json_data, headers=self._auth_header)
#Update a record
#Parameters
#----------
#module : Module
# The module you want to get the data for.
#id : string
# The if of the object to be changed
#attributes : dict[str, object]
# The attributes to be updated
def update_data(self, module, id, attributes) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
if id is None or id == "":
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("id"))
if attributes is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("attributes"))
if not isinstance(attributes, dict):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("attributes", "dict"))
body = {'data': { 'type': module.name.lower().capitalize(), 'id': id, 'attributes': attributes}}
json_data = json.dumps(body, default=lambda o: o.__dict__, sort_keys=False)
return requests.patch('{0}/Api/V8/module'.format(self._host), data = json_data, headers=self._auth_header)
#Get data by a given id
#Parameters
#----------
#module : string
# The name of the module you want to get the data for. You can use the constants defined in constants.py
#id : string
# The if of the data (object) to get
#fields : list
# The fields to be returned. If None then all fields will be returned.
def get_data_by_id(self, module, id, fields=None) -> requests.Response:
if module is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("module"))
if not isinstance(module, Module):
raise TypeError(self.PARAMETER_MUST_BE_OF_TYPE.format("module", "enum Module"))
if id is None:
raise TypeError(self.PARAMETER_MUST_NOT_BE_NONE.format("id"))
seperator = ","
if module != None and type(fields) == list and len(fields) > 0:
fields = "fields[{0}]={1}".format(module.value, seperator.join(fields))
response = requests.get("{0}/Api/V8/module/{1}/{2}{3}".
format(self._host, module.value, id, self._build_query_params(fields, None, None)),
headers=self._auth_header)
return response
#Build the query parameters to be appended to the API call
#Parameters
#----------
#fields : list
# The fields to be returned. If None then all fields will be returned.
#filter : Filter
# The filter you want ot set. Default is None
#pagination: Pagination
# The pagination settings to be set. Default page size is 50
def _build_query_params(self, fields, filter, pagination) -> str:
connectors = ["?", "&"]
query_string = ""
connector_index = 0
if fields != None and len(fields) > 0:
query_string += "{0}{1}".format(connectors[connector_index], fields)
connector_index = 1
if filter != None:
query_string += "{0}{1}".format(connectors[connector_index], filter.to_filter_string())
connector_index = 1
if pagination != None:
if pagination.page_number != None:
pages = "page[number]={0}&page[size]={1}".format(pagination.page_number, pagination.page_size)
else:
pages = "page[size]={0}".format(pagination.page_size)
query_string += "{0}{1}".format(connectors[connector_index], pages)
connector_index = 1
return query_string