|
from typing import List
from django.contrib.auth.models import AbstractUser
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models import Q
from django.utils import timezone
from django.utils.functional import cached_property
from rest_framework.response import Response
from backend_app.models.abstract.base import BaseModelSerializer, BaseModel
from backend_app.models.abstract.base import BaseModelViewSet
from backend_app.permissions.app_permissions import IsOwner, ReadOnly
from backend_app.utils import get_user_level, OBJ_MODERATION_PERMISSIONS
def validate(user, allow_sharing_personal_info):
"""
Custom validation to ensure that moderators, DRI and staff can't be "anonymous" on the plateform
"""
if (
get_user_level(user) >= OBJ_MODERATION_PERMISSIONS["moderator"]
and not allow_sharing_personal_info
):
raise ValidationError(
{
"allow_sharing_personal_info": "Users that are moderators, members of DRI or staff, must allow sharing of their 'identity', sorry."
}
)
class User(AbstractUser):
"""
If you modify this model, don't forget to modify the command that handles its emptying for RGPD
"""
@cached_property
def cached_groups(self) -> List[str]:
out = ["authenticated_user"]
for group in self.groups.all():
out.append(group.name)
return out
allow_sharing_personal_info = models.BooleanField(default=True, null=False)
secondary_email = models.EmailField(null=True, blank=True)
pseudo = models.CharField(
blank=False, null=False, max_length=12, default="Anonymous42"
)
has_validated_cgu_rgpd = models.BooleanField(default=False, null=False)
is_banned = models.BooleanField(default=False, null=False)
# Handling of account deletion
delete_next_time = models.BooleanField(
default=False, null=False
) # if true, the account will be deleted at midnight
is_deleted = models.BooleanField(default=False, null=False)
def save(self, *args, **kwargs):
"""
Custom save function to ensure consistency.
"""
# if the object is not created yet, we can't check to what groups it belongs
if self.pk is not None:
validate(self, self.allow_sharing_personal_info)
return super().save(*args, **kwargs)
class UserSerializer(BaseModelSerializer):
def validate(self, attrs):
"""
Also validate at the serializer level to prevent error 500
"""
data = super().validate(attrs)
aspi = data["allow_sharing_personal_info"]
validate(self.get_user_from_request(), aspi)
return data
class Meta:
model = User
fields = BaseModelSerializer.Meta.fields + (
"username",
"first_name",
"last_name",
"email",
"allow_sharing_personal_info",
"secondary_email",
"pseudo",
"delete_next_time",
"is_staff",
)
read_only_fields = ("username", "first_name", "last_name", "email")
class UserViewset(BaseModelViewSet):
def list(self, request, *args, **kwargs):
# Prevent the querying of all objects.
return Response(list())
def retrieve(self, request, *args, **kwargs):
# Custom behavior to return only the correct set of attributes depending
# On allow_sharing_personal_info
instance = self.get_object()
serializer = self.get_serializer(instance)
if instance == self.request.user or self.request.user.is_staff:
out = serializer.data
else:
# serializer.data is a property we can't set it again, so we need a little trick for the output
# First we copy all the values
out = dict(serializer.data)
# Then we "correct" them
for key in [
"username",
"first_name",
"last_name",
"email",
"secondary_email",
]:
out[key] = None
return Response(out)
queryset = User.objects.all()
permission_classes = (IsOwner | ReadOnly,)
serializer_class = UserSerializer
end_point_route = "users"
SITE_INFORMATION_VARIANTS = ["info", "success", "warning", "error"]
class SiteInformation(BaseModel):
start = models.DateTimeField(null=False)
end = models.DateTimeField(null=True)
message = models.TextField(max_length=500)
variant = models.CharField(
max_length=10,
choices=[(c, c) for c in SITE_INFORMATION_VARIANTS],
default="info",
null=False,
blank=False,
)
should_notify = models.BooleanField(default=False, null=False)
class SiteInformationSerializer(BaseModelSerializer):
class Meta:
model = SiteInformation
fields = BaseModelSerializer.Meta.fields + (
"start",
"end",
"message",
"variant",
"should_notify",
)
class SiteInformationViewSet(BaseModelViewSet):
permission_classes = (ReadOnly,)
serializer_class = SiteInformationSerializer
end_point_route = "information"
def get_queryset(self):
get = self.request.GET
if "now" in get.keys():
now = timezone.now()
return SiteInformation.objects.filter(
Q(start__lte=now) & (Q(end__isnull=True) | Q(end__gte=now))
)
else:
return SiteInformation.objects.all()
|
