|
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).order_by("revision_id")
|
