Initial commit
This commit is contained in:
45
webservice/lib/api/base.py
Normal file
45
webservice/lib/api/base.py
Normal file
@@ -0,0 +1,45 @@
|
||||
from rest_framework import viewsets, permissions, serializers
|
||||
from rest_framework.permissions import BasePermission
|
||||
|
||||
class IsOwner(BasePermission):
|
||||
def has_object_permission (self, request, view, obj ):
|
||||
"""Return 'True' if permission is granted, 'False' otherwise."""
|
||||
# TODO: If this is the 'way to go', we should consider adding the researcher reference to all models and save actions
|
||||
return obj.researcher == request.user.researcher or obj.study.researcher == request.user.researcher
|
||||
|
||||
class BaseReadOnlyViewSet(viewsets.ReadOnlyModelViewSet):
|
||||
permission_classes = [permissions.IsAuthenticated, IsOwner]
|
||||
|
||||
# TODO: If this is the 'way to go', we should consider adding the researcher reference to all models and save actions
|
||||
def get_queryset(self):
|
||||
try:
|
||||
qs = self.queryset.filter(researcher = self.request.user.researcher)
|
||||
except:
|
||||
qs = self.queryset.filter(study__researcher = self.request.user.researcher)
|
||||
|
||||
return qs
|
||||
|
||||
class BaseViewSet(viewsets.ModelViewSet):
|
||||
permission_classes = [permissions.IsAuthenticated, IsOwner]
|
||||
|
||||
# TODO: If this is the 'way to go', we should consider adding the researcher reference to all models and save actions
|
||||
def get_queryset(self):
|
||||
try:
|
||||
qs = self.queryset.filter(researcher = self.request.user.researcher)
|
||||
except:
|
||||
qs = self.queryset.filter(study__researcher = self.request.user.researcher)
|
||||
|
||||
return qs
|
||||
|
||||
class BaseHyperlinkedModelSerializer(serializers.HyperlinkedModelSerializer):
|
||||
# This ID field is handy to have.... Due to HyperlinkedModelSerializer we do not have this field by default
|
||||
id = serializers.ReadOnlyField()
|
||||
|
||||
# Only show the researcher full name
|
||||
researcher = serializers.StringRelatedField()
|
||||
|
||||
# Only show link to full researcher data
|
||||
#researcher = serializers.HyperlinkedRelatedField(view_name= 'api:v1:researcher-detail', read_only=True)
|
||||
|
||||
# Show the full researcher information
|
||||
#researcher = ResearcherSerializer(read_only=True)
|
||||
145
webservice/lib/api/client.py
Normal file
145
webservice/lib/api/client.py
Normal file
@@ -0,0 +1,145 @@
|
||||
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
|
||||
Reference in New Issue
Block a user