Repository URL to install this package:
|
Version:
1.1.1 ▾
|
import inspect, warnings
from rest_framework.response import Response
from rest_framework import status
from django_fsm import TransitionNotAllowed
from .exceptions import ChangeDataJSONParseException, ChangeInstanceForbiddenException, DeleteInstanceForbiddenException
from .permissions import ProtectDataPermission
def detail_fsm_route(obj, action, request, serializer_class, delete_instance=False):
''' Routes the Finate State Machine through the Rest Framework '''
warnings.warn(
f"{inspect.stack()[0].function} is going to be depecated in version 1.1.0. Please make sure to use the FSMViewMixin instead.",
PendingDeprecationWarning
)
if request.method == 'GET':
return Response(serializer_class(instance=obj, context={'request':request}).data)
serializer = serializer_class(instance=obj, data=request.data, partial=True, context={'request':request})
if serializer.is_valid():
errors = None
obj = serializer.save()
if hasattr(obj, f'can_{action}'):
errors = (getattr(obj, f'can_{action}')())
if errors is None or len(errors.keys()) == 0:
getattr(obj, action)(by=request.user, description=request.data.get('fsm_description', ''))
obj.save()
if delete_instance:
obj.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
serializer = serializer_class(instance=obj, context={'request':request})
return Response(serializer.data)
return Response(errors, status=status.HTTP_412_PRECONDITION_FAILED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class FSMViewMixin:
"""Implements a method to call in order to make FSM Routing more easy
"""
def get_permissions(self):
permissions = super().get_permissions()
permissions.append(ProtectDataPermission())
return permissions
def handle_exception(self, exc):
if type(exc) in [ChangeInstanceForbiddenException, DeleteInstanceForbiddenException]:
return Response(
{'non_field_errors': str(exc)},
status=status.HTTP_403_FORBIDDEN
)
if type(exc) == ChangeDataJSONParseException:
return Response(
{'change_data': str(exc)},
status=status.HTTP_400_BAD_REQUEST
)
if type(exc) == TransitionNotAllowed:
return Response(
{'non_field_errors': str(exc)},
status=status.HTTP_400_BAD_REQUEST
)
return super().handle_exception(exc)
def fsm_route(self, request):
obj = self.get_object()
serializer_class = self.get_serializer_class()
action = inspect.stack()[1].function
if request.method == 'GET':
return Response(serializer_class(instance=obj, context={'request':request}).data)
serializer = serializer_class(instance=obj, data=request.data, partial=True, context={'request':request})
if serializer.is_valid():
if len(serializer.validated_data) > 0:
obj = serializer.save()
else:
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
errors = None
if hasattr(obj, f'can_{action}'):
errors = (getattr(obj, f'can_{action}')())
if errors is None or len(errors.keys()) == 0:
data = None
# Check if changed_data is a valid attribut and has data
if hasattr(obj, 'change_data') and obj.change_data is not None and len(obj.change_data.keys()) > 0:
data = dict()
original_obj = obj.__class__.objects.get(id=obj.id)
for k, v in obj.change_data.items():
data[k] = getattr(original_obj, k)
getattr(obj, action)(by=request.user, description=request.data.get('fsm_description', ''), data=data)
obj.save()
serializer = serializer_class(instance=obj, context={'request':request})
return Response(serializer.data)
return Response(errors, status=status.HTTP_412_PRECONDITION_FAILED)