Skip to content

Instantly share code, notes, and snippets.

@floweb

floweb/cyclos_api.py

Created February 2, 2017 08:35
Show Gist options
  • Save floweb/b4fa9b97ebf201eb002910abd010f438 to your computer and use it in GitHub Desktop.
Save floweb/b4fa9b97ebf201eb002910abd010f438 to your computer and use it in GitHub Desktop.
Download ZIP
API wrapper examples
Raw
cyclos_api.py
import logging
from django.conf import settings
from rest_framework.exceptions import APIException
import requests
log = logging.getLogger()
class CyclosAPIException(APIException):
status_code = 400
default_detail = 'Cyclos API Exception'
class CyclosAPI(object):
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
try:
self.url
except AttributeError:
self.url = settings.CYCLOS_URL
try:
if self.mode == 'bdc':
self._init_bdc()
elif self.mode == 'cel':
self._init_cel()
elif self.mode == 'gi':
self._init_gi()
elif self.mode == 'gi_bdc' and self.login_bdc:
self._init_gi_bdc()
except AttributeError:
pass
def _init_bdc(self):
# getCurrentUser => get ID for current user
try:
self.user_profile = self.post(method='user/getCurrentUser', data=[])
self.user_id = self.user_profile['result']['id']
except CyclosAPIException:
raise CyclosAPIException(detail='Unable to connect to Cyclos!')
except KeyError:
raise CyclosAPIException(detail='Unable to fetch Cyclos data! Maybe your credentials are invalid!?')
# user/load for this ID to get field BDC ID
try:
self.user_data = self.post(method='user/load', data=self.user_id)
self.user_bdc_id = [item['linkedEntityValue']['id']
for item in self.user_data['result']['customValues']
if item['field']['id'] ==
str(settings.CYCLOS_CONSTANTS['user_custom_fields']['bdc'])][0]
except CyclosAPIException:
raise CyclosAPIException(detail='Unable to connect to Cyclos!')
except (KeyError, IndexError):
raise CyclosAPIException(detail='Unable to fetch Cyclos data! Maybe your credentials are invalid!?')
def _init_cel(self):
try:
self.user_profile = self.post(method='user/getCurrentUser', data=[])
self.user_id = self.user_profile['result']['id']
except CyclosAPIException:
raise CyclosAPIException(detail='Unable to connect to Cyclos!')
except KeyError:
raise CyclosAPIException(detail='Unable to fetch Cyclos data! Maybe your credentials are invalid!?')
def _init_gi(self):
# getCurrentUser => get ID for current user
try:
self.user_profile = self.post(method='user/getCurrentUser', data=[])
self.user_id = self.user_profile['result']['id']
except CyclosAPIException:
raise CyclosAPIException(detail='Unable to connect to Cyclos!')
except KeyError:
raise CyclosAPIException(detail='Unable to fetch Cyclos data! Maybe your credentials are invalid!?')
def _init_gi_bdc(self):
self.user_id = self.get_member_id_from_login(self.login_bdc)
self.user_bdc_id = self.user_id
def get_member_id_from_login(self, member_login, auth_string=None):
if auth_string:
self._handle_auth_string(auth_string)
query_data = {
'keywords': member_login,
'userStatus': ['ACTIVE', 'BLOCKED', 'DISABLED']
}
try:
member_login_data = self.post(method='user/search', data=query_data)
member_cyclos_id = member_login_data['result']['pageItems'][0]['id']
except CyclosAPIException:
raise CyclosAPIException(detail='Unable to connect to Cyclos!')
except (KeyError, IndexError):
raise CyclosAPIException(detail='Unable to fetch Cyclos data! Maybe your credentials are invalid!?')
return member_cyclos_id
def _handle_auth_string(self, auth_string):
log.debug(auth_string)
self.auth_string = auth_string
return auth_string
def _handle_auth_headers(self, headers):
headers.update({'Authorization': 'Basic {}'.format(self.auth_string)})
return headers
def _handle_api_response(self, api_response):
""" In some cases, we have to deal with errors in the response from the cyclos api !
"""
if api_response.status_code == requests.codes.ok:
response_data = api_response.json()
else:
message = 'Cyclos API Exception in {} - {}: {} - {}'.format(
api_response.request.method, api_response.url, api_response, api_response.text)
log.critical(message)
raise CyclosAPIException(detail=message)
# We don't have errors in our response, we can go on... and handle the response in our view.
log.info("response_data for {} - {}: {}".format(api_response.request.method, api_response.url, response_data))
return response_data
def get(self, method, id=None, auth_string=None, **kwargs):
if auth_string:
self._handle_auth_string(auth_string)
if id:
query = '{}/{}/{}'.format(self.url, method, id)
else:
query = '{}/{}'.format(self.url, method)
for key, value in kwargs.items():
query = "{}&{}={}".format(query, key, value)
r = requests.get(query, headers=self._handle_auth_headers({'content-type': 'application/json'}))
return self._handle_api_response(r)
def post(self, method, data, id=None, auth_string=None):
if auth_string:
self._handle_auth_string(auth_string)
if id:
query = '{}/{}/{}'.format(self.url, method, id)
else:
query = '{}/{}'.format(self.url, method)
r = requests.post(query, json=data, headers=self._handle_auth_headers({'content-type': 'application/json'}))
return self._handle_api_response(r)
def patch(self, method, data, id=None, auth_string=None):
if auth_string:
self._handle_auth_string(auth_string)
if id:
query = '{}/{}/{}'.format(self.url, method, id)
else:
query = '{}/{}'.format(self.url, method)
r = requests.patch(query, json=data, headers=self._handle_auth_headers({'content-type': 'application/json'}))
return self._handle_api_response(r)
def delete(self, method, id=None, auth_string=None):
if auth_string:
self._handle_auth_string(auth_string)
if id:
query = '{}/{}/{}'.format(self.url, method, id)
else:
query = '{}/{}'.format(self.url, method)
r = requests.delete(query, headers=self._handle_auth_headers({'content-type': 'application/json'}))
return self._handle_api_response(r)
Raw
dolibarr_api.py
import logging
from django.conf import settings
from rest_framework.exceptions import APIException
import requests
log = logging.getLogger()
class DolibarrAPIException(APIException):
status_code = 400
default_detail = 'Dolibarr API Exception'
class DolibarrAPI(object):
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
try:
self.url
except AttributeError:
self.url = settings.DOLIBARR_URL
def _handle_api_key(self, api_key):
log.debug(api_key)
self.api_key = api_key
return api_key
def _handle_api_response(self, api_response):
""" In some cases, we have to deal with errors in the response from the dolibarr api !
"""
if api_response.status_code == requests.codes.ok:
response_data = api_response.json()
else:
message = 'Dolibarr API Exception in {} - {}: {} - {}'.format(
api_response.request.method, api_response.url, api_response, api_response.text)
log.critical(message)
raise DolibarrAPIException(detail=message)
# We don't have errors in our response, we can go on... and handle the response in our view.
log.info("response_data for {} - {}: {}".format(api_response.request.method, api_response.url, response_data))
return response_data
def login(self, login=None, password=None):
""" Login function for Dolibarr API users. """
r = requests.get('{}/login?login={}&password={}'.format(self.url, login, password),
headers={'content-type': 'application/json'})
json_response = r.json()
if r.status_code == requests.codes.ok:
try:
api_key = json_response['success']['token']
except KeyError:
try:
message = 'Dolibarr API Exception: {}'.format(json_response['errors']['message'])
raise DolibarrAPIException(status_code=json_response['errors']['code'], detail=message)
except KeyError:
raise DolibarrAPIException()
else:
try:
message = 'Dolibarr API Exception: {}'.format(json_response['errors']['message'])
raise DolibarrAPIException(status_code=json_response['errors']['code'], detail=message)
except KeyError:
raise DolibarrAPIException()
return self._handle_api_key(api_key)
def get(self, model, id=None, api_key=None, **kwargs):
if api_key:
self._handle_api_key(api_key)
if id:
query = '{}/{}/{}?api_key={}'.format(self.url, model, id, self.api_key)
else:
query = '{}/{}?api_key={}'.format(self.url, model, self.api_key)
for key, value in kwargs.items():
query = "{}&{}={}".format(query, key, value)
r = requests.get(query, headers={'content-type': 'application/json'})
return self._handle_api_response(r)
def post(self, model, data, id=None, api_key=None):
if api_key:
self._handle_api_key(api_key)
if id:
query = '{}/{}/{}?api_key={}'.format(self.url, model, id, self.api_key)
else:
query = '{}/{}?api_key={}'.format(self.url, model, self.api_key)
r = requests.post(query, json=data, headers={'content-type': 'application/json'})
return self._handle_api_response(r)
def put(self, model, data, id=None, api_key=None):
if api_key:
self._handle_api_key(api_key)
if id:
query = '{}/{}/{}?api_key={}'.format(self.url, model, id, self.api_key)
else:
query = '{}/{}?api_key={}'.format(self.url, model, self.api_key)
r = requests.put(query, json=data, headers={'content-type': 'application/json'})
return self._handle_api_response(r)
def patch(self, model, data, id=None, api_key=None):
if api_key:
self._handle_api_key(api_key)
if id:
query = '{}/{}/{}?api_key={}'.format(self.url, model, id, self.api_key)
else:
query = '{}/{}?api_key={}'.format(self.url, model, self.api_key)
r = requests.patch(query, json=data, headers={'content-type': 'application/json'})
return self._handle_api_response(r)
def delete(self, model, id=None, api_key=None):
if api_key:
self._handle_api_key(api_key)
if id:
query = '{}/{}/{}?api_key={}'.format(self.url, model, id, self.api_key)
else:
query = '{}/{}?api_key={}'.format(self.url, model, self.api_key)
r = requests.delete(query, headers={'content-type': 'application/json'})
return self._handle_api_response(r)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment