from django.contrib.contenttypes.models import ContentType from django.utils import timezone from rest_framework import serializers from backend_app.permissions.moderation import is_moderation_required from backend_app.permissions.utils import Request as FakeRequest from backend_app.utils import get_user_level from base_app.models import User from django.contrib.contenttypes.fields import GenericRelation from django.core.exceptions import ValidationError from django.core.validators import MinValueValidator from django.db import models from backend_app.settings.defaults import ( DEFAULT_OBJ_MODERATION_LV, OBJ_MODERATION_PERMISSIONS, ) from backend_app.models.pendingModeration import PendingModeration from .base import BaseModel, BaseModelSerializer, BaseModelViewSet POSSIBLE_OBJ_MODER_LV = [ OBJ_MODERATION_PERMISSIONS[key] for key in OBJ_MODERATION_PERMISSIONS ] def validate_obj_model_lv(value): if value not in POSSIBLE_OBJ_MODER_LV: raise ValidationError("obj_moderation_level not recognized") # # # # # # # Module # # # # # # # # class EssentialModule(BaseModel): """ All models in the app deppend of this one. It contains the required attributes for managing optional data moderation. All the logic behind moderation is done in EssentialModuleSerializer """ # store the update author updated_by = models.ForeignKey( User, null=True, on_delete=models.SET_NULL, related_name="+" ) # store the update date (model can be updated without moderation) updated_on = models.DateTimeField(null=True) # store the moderator moderated_by = models.ForeignKey( User, null=True, on_delete=models.SET_NULL, related_name="+" ) # store the moderation date moderated_on = models.DateTimeField(null=True) # Store the object moderation level by default obj_moderation_level = models.SmallIntegerField( default=DEFAULT_OBJ_MODERATION_LV, validators=[MinValueValidator(0), validate_obj_model_lv], ) # Add the link to pending moderation pending_moderation = GenericRelation(PendingModeration) # A bit of optimization: we store if there is something pending moderation has_pending_moderation = models.BooleanField(default=False) class Meta: abstract = True # # # # # # # Serializer # # # # # # # # CLEANED_ESSENTIAL_MODULE_MODEL_DATA = { "moderated_by": None, "moderated_on": None, "updated_by": None, "updated_on": None, } def override_data(old_data: dict, new_data: dict) -> dict: """Update the data in old_data with the one in new_data """ for key in new_data: if key in old_data: old_data[key] = new_data[key] return old_data class EssentialModuleSerializer(BaseModelSerializer): """Serializer to go along the EssentialModule Model. This serializer handles backend data moderation checks and tricks. Raises: ValidationError -- If you are trying to moderate something you don't have rights to """ ###### # Basic fields serializers updated_by = serializers.CharField(read_only=True) updated_on = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", read_only=True) moderated_by = serializers.CharField(read_only=True) moderated_on = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", read_only=True) has_pending_moderation = serializers.BooleanField(read_only=True) # For easier handling on the client side, we force an id field # this is useful when a model has a dedicated primary key id = serializers.SerializerMethodField() # Add a content_type_id field to be able to find versions content_type_id = serializers.SerializerMethodField() def get_content_type_id(self, obj): """ Serializer for content type """ return ContentType.objects.get_for_model(self.Meta.model).id def get_obj_info(self, obj) -> dict: """ Serializer for the `obj_info` *dynamic* field. """ try: user_can_edit = self.context["user_can_edit"] except KeyError: # In case some viewsets don't inherit from BaseModelViewSet and therefore # don't have the method to produce context["user_can_edit"] # Anyway, those Viewsets should be readonly, so we can return false. user_can_edit = False return { "user_can_edit": user_can_edit, "user_can_moderate": not is_moderation_required( self.Meta.model, obj, self.get_user_from_request() ), } class Meta: model = EssentialModule def get_user_from_request(self): """ Function to retrieve the user from the request """ return self.context["request"].user def validate(self, attrs): """ Validate `BaseModel` fields and enforce certain field at the backend level. Checks that the requested moderation level is not higher than the one of the user. """ if "obj_moderation_level" in attrs: requested_obj_moder_lv = attrs["obj_moderation_level"] try: user_level = get_user_level(self.get_user_from_request()) except KeyError: # if for some reason we don't have the user in the request # we set the level to the default one # this can occur during testing. user_level = DEFAULT_OBJ_MODERATION_LV if requested_obj_moder_lv > user_level: raise ValidationError( "You can't request moderation for a higher rank than you." ) return attrs def set_model_attrs_for_moderation_and_update( self, user, moderated_and_updated: bool ): """ Overrides model attributes regarding moderation and update. The moderated field is set to the request user. The moderated_on field is reset to now. If there was an updated, the updated_by field and updated_on field are also reset. """ now = timezone.now() self.override_validated_data({"moderated_by": user, "moderated_on": now}) if moderated_and_updated: self.override_validated_data({"updated_by": user, "updated_on": now}) def clean_validated_data(self): """ Clear fields related to update and moderation """ self.override_validated_data(CLEANED_ESSENTIAL_MODULE_MODEL_DATA) def override_validated_data(self, new_data: dict): """ Method used to force specific attributes when saving a model """ for key in new_data: self.validated_data[key] = new_data[key] def do_before_save(self): """ Action to perform before saving a model """ pass def save(self, *args, **kwargs): """ Function that handles all the moderation in a smart way. Nothing has to be done to tell that we won't the data to be moderated, it is detected automatically. """ user = self.context["request"].user user_level = get_user_level(user) self.clean_validated_data() self.do_before_save() ct = ContentType.objects.get_for_model(self.Meta.model) if is_moderation_required(self.Meta.model, self.instance, user, user_level): if self.instance is None: # we need to create the main model # Store the user for squashing data in versions models self.validated_data.updated_by = user self.instance = super().save(*args, **kwargs) data_to_save = dict() for key in self.validated_data: try: # retrieve the submitted data and save the clean json # to make sure it will be savable data_to_save[key] = self.initial_data[key] except KeyError: pass data_to_save = override_data( data_to_save, CLEANED_ESSENTIAL_MODULE_MODEL_DATA ) # Save instance into pending moderation state PendingModeration.objects.update_or_create( content_type=ct, object_id=self.instance.pk, defaults={ "updated_on": timezone.now(), "updated_by": user, "new_object": data_to_save, }, ) # Performance optimization, we store the fact that there is an object pending moderation self.instance.has_pending_moderation = True self.instance.save() return self.instance else: # Moderation is not needed, we need to check whether it's a moderation or an update with no moderation moderated_and_updated = True if self.instance is None: self.set_model_attrs_for_moderation_and_update( user, moderated_and_updated ) instance = super().save(*args, **kwargs) else: try: pending_instance = PendingModeration.objects.get( content_type=ct, object_id=self.instance.pk ) self.clean_validated_data() # Make quez that it is done... # We have to compare the serialized data # So we make sure to compare the same elements key_to_remove = [] for key in self.initial_data: if key not in self.validated_data: key_to_remove.append(key) for key in key_to_remove: self.initial_data.pop(key, None) if pending_instance.new_object == self.initial_data: moderated_and_updated = False self.validated_data["updated_by"] = pending_instance.updated_by self.validated_data["updated_on"] = pending_instance.updated_on pending_instance.delete() except PendingModeration.DoesNotExist: pass self.set_model_attrs_for_moderation_and_update( user, moderated_and_updated ) instance = super().save(*args, **kwargs) # Performance optimization to know if has pending moderation instance.has_pending_moderation = False instance.save() return instance # # # # # # # ViewSet # # # # # # # # class EssentialModuleViewSet(BaseModelViewSet): """ Custom default viewset """ serializer_class = EssentialModuleSerializer def get_serializer_context(self): """ Override default function. Extra context is provided to the serializer class to know if a user can edit an element or not. This allows to not do this query for all elements and improves performances. You can look at the comment below for more information. """ # When generating the API documentation (url: /api-doc) the request would be None # and we don't need to do anything special if self.request is None: return super().get_serializer_context() fake_edit_request = FakeRequest(self.request.user, "PUT") user_can_edit = True for permission_class in self.get_permissions(): # Theoretically speaking we would need to use has_object_permission # But for performance purpose, we will consider edition right at the model # level. Which is consistent with our design. # Beware, that this might provide inconsistent data to the frontend # especially if permission_classes impact at the object level such as # IsOwner. if not permission_class.has_permission(fake_edit_request, None): user_can_edit = False break default_context = super().get_serializer_context() default_context["user_can_edit"] = user_can_edit return default_context def get_queryset(self): """ Extended default rest framework behavior to prefetch some table and enhance performances """ return self.queryset.prefetch_related("moderated_by", "updated_by")