version.py

from django.contrib.contenttypes.models import ContentType
from django.core import serializers as djangoSerializers
from django.core.serializers.base import DeserializationError
from rest_framework import serializers
from reversion.models import Version

from backend_app.models.abstract.base import BaseModelSerializer, BaseModelViewSet
from backend_app.permissions.app_permissions import ReadOnly


class VersionSerializer(BaseModelSerializer):
    """
    Custom serializer for the (reversion) version model
    """

    data = serializers.SerializerMethodField()

    serializers_mapping = None

    @classmethod
    def get_serializers_mapping(cls) -> dict:
        """
        Function that returns a mapping from model name to the serializer
        class that should be used to return the versioned data.
        """
        if cls.serializers_mapping is None:
            # Prevent cyclic imports
            from backend_app.viewsets import ALL_API_VIEWSETS

            # A little bit of optimization to easily find the serializer class associated with a model
            cls.serializers_mapping = dict()

            for viewset in ALL_API_VIEWSETS:
                serializer = viewset().get_serializer_class()
                model = serializer.Meta.model
                cls.serializers_mapping[model.__name__] = serializer

            # Override if models has a get_serializer method
            for viewset in ALL_API_VIEWSETS:
                model = viewset().get_serializer_class().Meta.model
                try:
                    cls.serializers_mapping[model.__name__] = model.get_serializer()
                except AttributeError:
                    pass
        return cls.serializers_mapping

    def get_data(self, obj):
        """
        Serializer for the data field
        """
        data = obj.serialized_data

        try:
            # We try to deserialize the version
            tmp = list(
                djangoSerializers.deserialize(obj.format, data, ignorenonexistent=True)
            )[0]
            # Version is valid,
            obj_serializer = self.get_serializers_mapping()[type(tmp.object).__name__]
            new_context = dict(self.context)
            new_context["view"].action = "list"
            return obj_serializer(tmp.object, context=new_context).data
        except (DeserializationError, djangoSerializers.SerializerDoesNotExist):
            # The version is not valid regarding the model, we should delete it !
            # This might be due to an updated model structure at some point.
            obj.delete()
            # We take care of the nb_versions field update with signals.
            # So it will remain coherent.
            return None

    class Meta:
        model = Version
        fields = ("data", "id")


class VersionViewSet(BaseModelViewSet):
    """
    Viewset for the versions
    """

    serializer_class = VersionSerializer
    permission_classes = (ReadOnly,)
    end_point_route = (
        r"versions/(?P<content_type_id>[0-9]+)/(?P<object_pk>[0-9A-Za-z]+)"
    )

    def get_queryset(self):
        content_type_id = self.kwargs["content_type_id"]
        object_pk = self.kwargs["object_pk"]
        ct = ContentType.objects.get_for_id(content_type_id)
        model = ct.model_class()
        obj = model.objects.get(pk=object_pk)
        return Version.objects.get_for_object(obj)