You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Adunatio/restapi/views.py

156 lines
5.0 KiB

import json
import time
import mongoengine
from flask import jsonify, request
from flask_views.base import View
from mongoengine import ValidationError, DoesNotExist, InvalidQueryError
from werkzeug.exceptions import Unauthorized, NotFound
from restapi.resource import Resource
class ApiView(View):
model = None
authentication_methods = []
def __init__(self, model):
self.start = time.time()
self.model = model
self.resource = Resource(self.model)
def dispatch_request(self, *args, **kwargs):
# keep all the logic in a helper method (_dispatch_request) so that
# it's easy for subclasses to override this method without them also having to copy/paste all the
# authentication logic, etc.
return self._dispatch_request(*args, **kwargs)
def _dispatch_request(self, *args, **kwargs):
authorized = True if len(self.authentication_methods) == 0 else False
for authentication_method in self.authentication_methods:
if authentication_method().authorized():
authorized = True
if not authorized:
return {'error': 'Unauthorized'}, '401 Unauthorized'
try:
return super(ApiView, self).dispatch_request(*args, **kwargs)
except mongoengine.queryset.DoesNotExist as e:
return {'error': 'Empty query: ' + str(e)}, '404 Not Found'
except ValidationError as e:
return e.args[0], '400 Bad Request'
except Unauthorized:
return {'error': 'Unauthorized'}, '401 Unauthorized'
except NotFound as e:
return {'error': str(e)}, '404 Not Found'
def get(self, *args, **kwargs):
"""
TODO: check permissions
:param args:
:param kwargs:
:return:
"""
if 'pk' in kwargs:
try:
qs = self.resource.to_qs(pk=kwargs.get('pk'))
count, data = self.resource.to_json(pk=kwargs.get('pk'))
response = {
'response': data,
'status': True,
'info': {
'took': time.time() - self.start
},
}
except (ValidationError, DoesNotExist) as e:
kwargs.update({'code': 404})
response = {
"status": False,
"errors": "Object not found"
}
else:
if "query" in request.args and self.model._meta.get('can_query'):
self.resource.external_query(json.loads(request.args.get('query')))
count, data = self.resource.to_json()
response = {
'response': data,
'info': {
'offset': self.resource.offset,
'limit': self.resource.limit,
'status': True,
'total': count,
'took': time.time() - self.start
},
}
return jsonify(response), kwargs.get('code', 200)
def post(self, *args, **kwargs):
"""
TODO: check permissions
:param args:
:param kwargs:
:return:
"""
try:
item = self.model(**request.json)
item.validate()
except ValidationError as v:
return jsonify({
'status': False,
'errors': [{"field": field, "error": str(error)} for field, error in v.__dict__.get('errors').items() if
isinstance(error, ValidationError) and field[0] != "_"]
}), 400
except Exception as e:
return jsonify({
'status': False,
'errors': str(e)
}), 400
data = item.save()
return self.get(pk=data.id, code=201)
def put(self, *args, **kwargs):
"""
TODO: check permissions
:param args:
:param kwargs:
:return:
"""
if "pk" not in kwargs:
return jsonify({
'status': False,
"error": "Method not allowed"
}), 403
else:
try:
self.model.objects(id=kwargs.get('pk')).update(**request.json)
return self.get(pk=kwargs.get('pk'))
except ValidationError as v:
print(v.__dict__)
return jsonify({
'status': False,
'errors': [{"field": v.__dict__.get('field_name'), "error": v.__dict__.get('_message')}]
}), 400
except InvalidQueryError as e:
return jsonify({
'status': False,
'errors': str(e)
}), 400
def has_read_permission(self, request, qs):
return qs
def has_add_permission(self, request, obj):
return True
def has_change_permission(self, request, obj):
return True
def has_delete_permission(self, request, obj):
return True
def delete(self, *args, **kwargs):
"delete method"