Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
wbfsm / wbfsm / views.py
Size: Mime:
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)