145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import requests
|
|
from requests_hawk import HawkAuth
|
|
from datetime import datetime
|
|
|
|
class VRE_API_Exception(BaseException):
|
|
def __init__(self, message, error):
|
|
# Call the base class constructor with the parameters it needs
|
|
super().__init__(message)
|
|
# Now for your custom code...
|
|
self.error = error
|
|
|
|
def __repr__(self):
|
|
return '{} ({})'.format(self.message, self.error)
|
|
|
|
class VRE_API_Exception_Factory(VRE_API_Exception):
|
|
|
|
def __new__(self, message, error_code):
|
|
if error_code == 400:
|
|
return VRE_API_400(message)
|
|
elif error_code == 403:
|
|
return VRE_API_403(message)
|
|
elif error_code == 404:
|
|
return VRE_API_404(message)
|
|
|
|
class VRE_API_400(VRE_API_Exception):
|
|
def __init__(self, message):
|
|
# Call the base class constructor with the parameters it needs
|
|
super().__init__(message, 403)
|
|
|
|
class VRE_API_403(VRE_API_Exception):
|
|
def __init__(self, message):
|
|
# Call the base class constructor with the parameters it needs
|
|
super().__init__(message, 403)
|
|
|
|
class VRE_API_404(VRE_API_Exception):
|
|
def __init__(self, message):
|
|
# Call the base class constructor with the parameters it needs
|
|
super().__init__(message, 404)
|
|
|
|
class VRE_API_Client():
|
|
|
|
DATE_TIME_FIELDS = 'created_at,updated_at,mail_sent'.split(',')
|
|
|
|
HEADERS = {
|
|
'Content-Type': 'application/json',
|
|
'cache-control': 'no-cache'
|
|
}
|
|
|
|
def __init__(self, host, url = None, token = None, secret = None):
|
|
self.host = host
|
|
self.url = url
|
|
self.token = token
|
|
self.secret = secret
|
|
|
|
self.data = {}
|
|
self.authentication = HawkAuth(id=self.token , key=self.secret)
|
|
|
|
def __get_full_url(self):
|
|
return '{}{}'.format(self.host, self.url)
|
|
|
|
def __parse_date_time_fields(self, data):
|
|
for item in data:
|
|
# TODO: Should provide better solution for this try/catch. For now it works
|
|
try:
|
|
if isinstance(item,list) or isinstance(item,dict):
|
|
self.__parse_date_time_fields(item)
|
|
|
|
elif isinstance(data[item],list) or isinstance(data[item],dict):
|
|
self.__parse_date_time_fields(data[item])
|
|
|
|
elif item in self.DATE_TIME_FIELDS and isinstance(data[item],str):
|
|
try:
|
|
data[item] = datetime.strptime(data[item],'%Y-%m-%dT%H:%M:%S.%fZ')
|
|
except Exception:
|
|
data[item] = datetime.strptime(data[item][::-1].replace(':','',1)[::-1].replace(' ','T'),'%Y-%m-%dT%H:%M:%S.%f%z')
|
|
except Exception:
|
|
pass
|
|
|
|
def __parse_data(self, start = None):
|
|
if len(self.DATE_TIME_FIELDS) > 0:
|
|
self.__parse_date_time_fields(self.data)
|
|
|
|
def set_url(self, url):
|
|
self.url = url
|
|
|
|
def set_token(self, token):
|
|
self.token = token
|
|
self.authentication = HawkAuth(id=self.token , key=self.secret)
|
|
|
|
def set_secret(self, secret):
|
|
self.secret = secret
|
|
self.authentication = HawkAuth(id=self.token , key=self.secret)
|
|
|
|
def get_data(self):
|
|
result = requests.get(self.__get_full_url(), auth=self.authentication, headers=self.HEADERS)
|
|
self.data['status_code'] = result.status_code
|
|
|
|
if result.status_code in [200,201]:
|
|
self.data = result.json()
|
|
self.__parse_data()
|
|
else:
|
|
print(result.json())
|
|
raise VRE_API_Exception_Factory('Error with url {}.'.format(self.url),result.status_code)
|
|
|
|
return self.data
|
|
|
|
def post_data(self, payload):
|
|
result = requests.post(self.__get_full_url(), json=payload, auth=self.authentication, headers=self.HEADERS)
|
|
self.data['status_code'] = result.status_code
|
|
|
|
if result.status_code in [200,201]:
|
|
self.data = result.json()
|
|
self.__parse_data()
|
|
else:
|
|
#print(result.content)
|
|
|
|
#print(result.text)
|
|
|
|
raise VRE_API_Exception_Factory('Error with url {}.'.format(self.url),result.status_code)
|
|
|
|
return self.data
|
|
|
|
def put_data(self, payload):
|
|
result = requests.put(self.__get_full_url(), json=payload, auth=self.authentication, headers=self.HEADERS)
|
|
self.data['status_code'] = result.status_code
|
|
|
|
if result.status_code in [200,201]:
|
|
self.data = result.json()
|
|
self.__parse_data()
|
|
else:
|
|
print(result.json())
|
|
raise VRE_API_Exception_Factory('Error with url {}.'.format(self.url),result.status_code)
|
|
|
|
return self.data
|
|
|
|
def delete_data(self):
|
|
try:
|
|
# Django HAWK has issues with a delete action. It needs/wants a content-type header, but there is no content.....
|
|
# https://github.com/kumar303/hawkrest/issues/46
|
|
result = requests.delete(self.__get_full_url(), auth=self.authentication, headers=self.HEADERS)
|
|
return result.status_code in [200,201,204]
|
|
except Exception:
|
|
raise VRE_API_Exception_Factory('Error with url {}.'.format(self.url),result.status_code)
|
|
|
|
return False |