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