from typing import Any, Dict
from django.conf import settings
from django.contrib.auth.password_validation import validate_password
from django.contrib.auth.tokens import default_token_generator
from django.utils.timezone import now
from rest_framework import serializers
from rest_framework_dataclasses.serializers import DataclassSerializer
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer, TokenRefreshSerializer
from api.users.mixins.serializers import (
BaseUserActivationSerializer,
EmailValidationMixin,
PasswordMatchValidationMixin,
)
from apps.users.models import Category, Profile, User
from services.users import AccessRefreshTokensDTO, ProfileService
from tasks.users import send_email_confirmation
[docs]
class CategorySerializer(serializers.ModelSerializer):
class Meta:
model = Category
fields = "__all__"
[docs]
class ProfileSerializer(serializers.ModelSerializer):
class Meta:
model = Profile
exclude = ["id", "user"]
extra_kwargs = {"birthdate": {"required": True}}
[docs]
class UserSerializer(PasswordMatchValidationMixin, serializers.ModelSerializer):
interests = serializers.PrimaryKeyRelatedField(
many=True, queryset=Category.objects.all(), required=True, allow_empty=True, allow_null=False, write_only=True
)
interests_list = CategorySerializer(source="interests", read_only=True, many=True)
password_repeat = serializers.CharField(write_only=True, required=True)
is_current_user_following = serializers.BooleanField(default=False, read_only=True)
avatar = serializers.CharField(read_only=True, source="profile.avatar")
[docs]
def validate_password(self, value):
validate_password(value)
return value
[docs]
def get_fields(self):
fields = super().get_fields()
request = self.context.get("request")
if request and self.instance:
fields["password"].read_only = True
del fields["password_repeat"]
return fields
[docs]
def update(self, instance, validated_data):
if "email" in validated_data and validated_data["email"] != instance.email:
if settings.IS_TESTING:
send_email_confirmation(user_pk=instance.pk, email=validated_data["email"])
else:
send_email_confirmation.delay(user_pk=instance.pk, email=validated_data["email"])
validated_data["email"] = instance.email
profile_data = validated_data.pop("profile", "")
if profile_data:
ProfileService(instance).update(**profile_data)
return super().update(instance, validated_data)
[docs]
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.is_owner = kwargs.pop("is_owner", False)
self.private_fields = {"email", "birthdate"}
super(UserSerializer, self).__init__(*args, **kwargs)
# Dynamically add fields from ProfileSerializer
profile_serializer = ProfileSerializer()
for field_name, field in profile_serializer.fields.items():
field.source = f"profile.{field_name}"
self.fields[field_name] = field
writeable_fields = (
"password",
"password_repeat",
"interests",
"birthdate",
"avatar",
"email",
"username",
"first_name",
"last_name",
"bio",
)
read_only_field_names = tuple(
field.name
for field in self.Meta.model._meta.get_fields()
if not hasattr(field, "attname") or field.attname not in writeable_fields
)
for field_name in read_only_field_names:
if field_name in self.fields:
self.fields[field_name].read_only = True
[docs]
def to_representation(self, instance):
data = super().to_representation(instance)
request = self.context.get("request")
if request and instance and not self.is_owner:
is_owner = request.user == instance
else:
is_owner = False or self.is_owner
if not is_owner:
for field in self.private_fields:
data.pop(field, None)
return data
class Meta:
model = User
exclude = ["is_passed_onboarding", "groups", "user_permissions", "is_deleted"]
extra_kwargs = {
"password": {"write_only": True},
"following": {"read_only": True},
"first_name": {"required": True},
"last_name": {"required": True},
}
[docs]
class ExtendedUserSerializer(UserSerializer):
likes_quantity = serializers.IntegerField(default=0, read_only=True)
followers_quantity = serializers.IntegerField(default=0, read_only=True)
following_quantity = serializers.IntegerField(default=0, read_only=True)
[docs]
class FollowingUserSerializer(UserSerializer):
following = UserSerializer(source="limited_following", many=True, read_only=True)
[docs]
class FollowingAndFollowersUserSerializer(UserSerializer):
following = UserSerializer(many=True, read_only=True)
followers = UserSerializer(many=True, read_only=True)
[docs]
class UserActivationSerializer(BaseUserActivationSerializer):
token = serializers.CharField()
[docs]
def to_internal_value(self, data):
internal_value = super().to_internal_value(data)
internal_value["user"] = self.validate_user_by_uid(internal_value["uid"], check_is_active=True)
return internal_value
[docs]
def validate(self, attrs: Dict) -> Dict:
user = attrs["user"]
token = attrs["token"]
if not default_token_generator.check_token(user, token):
raise serializers.ValidationError({"token": "This token is invalid"})
return attrs
[docs]
class UserChangePasswordSerializer(PasswordMatchValidationMixin, serializers.Serializer):
password = serializers.CharField(write_only=True)
password_repeat = serializers.CharField(write_only=True)
[docs]
def validate_password(self, value):
validate_password(value)
return value
[docs]
class UserResetPassword(EmailValidationMixin, serializers.Serializer):
email = serializers.EmailField()
[docs]
class ConfirmEmailSerializer(BaseUserActivationSerializer):
token = serializers.CharField()
[docs]
def to_internal_value(self, data):
internal_value = super().to_internal_value(data)
internal_value["user"] = self.validate_user_by_uid(internal_value["uid"])
return internal_value
[docs]
def validate(self, attrs: Dict) -> Dict:
super().validate(attrs)
user = attrs["user"]
try:
token = attrs["token"]
except Exception:
raise serializers.ValidationError({"token": "Token is invalid"})
if not default_token_generator.check_token(user, token):
raise serializers.ValidationError({"token": "This token is invalid"})
activation_codes = user.activation_codes.filter(uid=attrs["uid"], code=token)
if not activation_codes.exists():
raise serializers.ValidationError({"activation_code": "This activation code doesn't belong this user"})
activation_code = activation_codes.first()
if now() > activation_code.expiration_date:
raise serializers.ValidationError({"token": "This token is expired"})
return attrs
[docs]
class ConfirmPasswordSerializer(ConfirmEmailSerializer, UserChangePasswordSerializer):
[docs]
def validate(self, attrs: Dict) -> Dict:
attrs = UserChangePasswordSerializer.validate(self, attrs)
attrs = ConfirmEmailSerializer.validate(self, attrs)
return attrs
[docs]
class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
user = FollowingUserSerializer(read_only=True, help_text="**!!! ONLY FOR RESPONSE !!!**")
[docs]
class CustomTokenRefreshSerializer(TokenRefreshSerializer):
user = FollowingUserSerializer(read_only=True, help_text="**!!! ONLY FOR RESPONSE !!!**")
[docs]
class AccessRefreshTokensDTOSerializer(DataclassSerializer):
class Meta:
dataclass = AccessRefreshTokensDTO
[docs]
class ResendUserActivationCodeSerializer(EmailValidationMixin, serializers.Serializer):
email = serializers.EmailField()
[docs]
def validate_email(self, value: str) -> str:
super().validate_email(value)
user = User.objects.filter(email=value, is_active=True, is_deleted=True)
if user.exists():
raise serializers.ValidationError("There is no inactive user with such email")
return value
[docs]
class CreateTokenDocsBodySerializer(serializers.Serializer):
email = serializers.EmailField(required=True)
password = serializers.CharField(required=True)
[docs]
class CreateTokenDocsResponseSerializer(serializers.Serializer):
user = FollowingUserSerializer()
access = serializers.CharField(help_text="Access JWT Token. Lifetime 5 min.")
refresh = serializers.CharField(help_text="Refresh JWT Token. Lifetime 30 days")
[docs]
class RefreshTokenDocsBodySerializer(serializers.Serializer):
refresh = serializers.CharField(required=True, help_text="Refresh JWT Token. Lifetime 30 days")
[docs]
class RefreshTokenDocsResponseSerializer(serializers.Serializer):
user = FollowingUserSerializer()
access = serializers.CharField(help_text="Access JWT Token. Lifetime 5 min.")
[docs]
class SenderNotificationSerializer(serializers.ModelSerializer):
"""Serializer for sender"""
avatar = serializers.CharField(read_only=True, source="profile.avatar")
is_current_user_following = serializers.BooleanField(default=False, read_only=True)
class Meta:
"""Meta-class Sender Notification"""
model = User
fields = ["id", "username", "is_current_user_following", "avatar"]