338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""Common account serializers"""
|
|
from django.conf import settings
|
|
from django.contrib.auth import password_validation as password_validators
|
|
from django.utils.translation import gettext_lazy as _
|
|
from fcm_django.models import FCMDevice
|
|
from rest_framework import exceptions
|
|
from rest_framework import serializers
|
|
from rest_framework import validators as rest_validators
|
|
|
|
from account import models, tasks
|
|
from account.models import User, Role
|
|
from main.serializers.common import NavigationBarPermissionBaseSerializer
|
|
from notification.models import Subscribe, Subscriber
|
|
from utils import exceptions as utils_exceptions
|
|
from utils import methods as utils_methods
|
|
from utils.methods import generate_string_code
|
|
from phonenumber_field.serializerfields import PhoneNumberField
|
|
|
|
|
|
def subscriptions_handler(subscriptions_list, user):
|
|
"""
|
|
create or update subscriptions for user
|
|
"""
|
|
Subscribe.objects.filter(subscriber__user=user).delete()
|
|
subscriber, _ = Subscriber.objects.get_or_create(
|
|
email=user.email,
|
|
defaults={
|
|
'user': user,
|
|
'email': user.email,
|
|
'ip_address': user.last_ip,
|
|
'country_code': user.last_country.country.code if user.last_country and user.last_country.country else None,
|
|
'locale': user.locale,
|
|
'update_code': generate_string_code(),
|
|
}
|
|
)
|
|
|
|
for subscription in subscriptions_list:
|
|
Subscribe.objects.create(
|
|
subscriber=subscriber,
|
|
subscription_type_id=subscription,
|
|
)
|
|
|
|
|
|
class RoleBaseSerializer(serializers.ModelSerializer):
|
|
"""Serializer for model Role."""
|
|
id = serializers.IntegerField()
|
|
role_display = serializers.CharField(source='get_role_display', read_only=True)
|
|
navigation_bar_permission = NavigationBarPermissionBaseSerializer(read_only=True)
|
|
country_code = serializers.CharField(source='country.code', read_only=True, allow_null=True)
|
|
country_name_translated = serializers.CharField(source='country.name_translated', read_only=True, allow_null=True)
|
|
|
|
class Meta:
|
|
"""Meta class."""
|
|
model = models.Role
|
|
fields = [
|
|
'id',
|
|
'role_display',
|
|
'navigation_bar_permission',
|
|
'country_code',
|
|
'country_name_translated',
|
|
]
|
|
|
|
|
|
class UserSerializer(serializers.ModelSerializer):
|
|
"""User serializer."""
|
|
# RESPONSE
|
|
fullname = serializers.CharField(source='get_full_name', read_only=True)
|
|
|
|
# REQUEST
|
|
username = serializers.CharField(
|
|
validators=(rest_validators.UniqueValidator(queryset=models.User.objects.all()),),
|
|
required=False)
|
|
email = serializers.EmailField(
|
|
validators=(rest_validators.UniqueValidator(queryset=models.User.objects.all()),),
|
|
required=False)
|
|
roles = RoleBaseSerializer(many=True, read_only=True)
|
|
subscriptions = serializers.ListField(
|
|
source='subscription_types',
|
|
allow_null=True,
|
|
allow_empty=True,
|
|
child=serializers.IntegerField(min_value=1),
|
|
required=False,
|
|
help_text='list of subscription_types id',
|
|
)
|
|
phone_as_international = PhoneNumberField(source='phone.as_international',
|
|
allow_null=True,
|
|
read_only=True)
|
|
|
|
class Meta:
|
|
model = models.User
|
|
fields = [
|
|
'id',
|
|
'username',
|
|
'first_name',
|
|
'last_name',
|
|
'fullname',
|
|
'cropped_image_url',
|
|
'image_url',
|
|
'email',
|
|
'email_confirmed',
|
|
'newsletter',
|
|
'roles',
|
|
'subscriptions',
|
|
'phone',
|
|
'phone_as_international',
|
|
]
|
|
extra_kwargs = {
|
|
'first_name': {'required': False, 'write_only': True, },
|
|
'last_name': {'required': False, 'write_only': True, },
|
|
'email_confirmed': {'read_only': True, },
|
|
'image_url': {'required': False, },
|
|
'cropped_image_url': {'required': False, },
|
|
'newsletter': {'required': False, },
|
|
'phone': {'required': False, 'write_only': True, }
|
|
}
|
|
|
|
def create(self, validated_data):
|
|
subscriptions_list = []
|
|
if 'subscription_types' in validated_data:
|
|
subscriptions_list = validated_data.pop('subscription_types')
|
|
|
|
user = super(UserSerializer, self).create(validated_data)
|
|
validated_data['user'] = user
|
|
Subscriber.objects.make_subscriber(**validated_data)
|
|
subscriptions_handler(subscriptions_list, user)
|
|
return user
|
|
|
|
def validate_username(self, value):
|
|
"""Custom username validation"""
|
|
valid = utils_methods.username_validator(username=value)
|
|
if not valid:
|
|
raise utils_exceptions.NotValidUsernameError()
|
|
return value
|
|
|
|
def update(self, instance, validated_data):
|
|
"""Override update method"""
|
|
subscriptions_list = []
|
|
|
|
if 'subscription_types' in validated_data:
|
|
subscriptions_list = validated_data.pop('subscription_types')
|
|
|
|
new_email = validated_data.get('email')
|
|
old_email = instance.email
|
|
instance = super().update(instance, validated_data)
|
|
request = self.context['request']
|
|
user = request.user
|
|
if not user.is_superuser and not user.is_staff and \
|
|
not user.roles.filter(country__code=request.country_code, role=models.Role.COUNTRY_ADMIN).exists():
|
|
"""
|
|
superuser or country admin changes email immediately!
|
|
"""
|
|
if new_email and new_email != old_email:
|
|
instance.email_confirmed = False
|
|
instance.email = old_email
|
|
instance.unconfirmed_email = new_email
|
|
instance.save()
|
|
# Send verification link on user email for change email address
|
|
if settings.USE_CELERY:
|
|
tasks.change_email_address.delay(
|
|
user_id=instance.id,
|
|
country_code=self.context.get('request').country_code,
|
|
emails=[validated_data['email'], ])
|
|
else:
|
|
tasks.change_email_address(
|
|
user_id=instance.id,
|
|
country_code=self.context.get('request').country_code,
|
|
emails=[validated_data['email'], ])
|
|
|
|
subscriptions_handler(subscriptions_list, instance)
|
|
|
|
return instance
|
|
|
|
|
|
class UserBaseSerializer(serializers.ModelSerializer):
|
|
"""Serializer is used to display brief information about the user."""
|
|
|
|
fullname = serializers.CharField(source='get_full_name', read_only=True)
|
|
|
|
class Meta:
|
|
"""Meta class."""
|
|
|
|
model = models.User
|
|
fields = (
|
|
'id',
|
|
'username',
|
|
'fullname',
|
|
'first_name',
|
|
'last_name',
|
|
'email',
|
|
'cropped_image_url',
|
|
'image_url',
|
|
)
|
|
read_only_fields = fields
|
|
|
|
|
|
class UserShortSerializer(UserSerializer):
|
|
"""Compact serializer for model User."""
|
|
|
|
class Meta(UserSerializer.Meta):
|
|
"""Meta class."""
|
|
fields = [
|
|
'id',
|
|
'fullname',
|
|
'email',
|
|
'username',
|
|
]
|
|
|
|
|
|
class ChangePasswordSerializer(serializers.ModelSerializer):
|
|
"""Serializer for model User."""
|
|
|
|
password = serializers.CharField(write_only=True)
|
|
old_password = serializers.CharField(write_only=True)
|
|
|
|
class Meta:
|
|
"""Meta class"""
|
|
model = models.User
|
|
fields = (
|
|
'password',
|
|
'old_password',
|
|
)
|
|
|
|
def validate(self, attrs):
|
|
"""Override validate method"""
|
|
password = attrs.get('password')
|
|
old_password = attrs.get('old_password')
|
|
try:
|
|
# Check old password
|
|
if not self.instance.check_password(raw_password=old_password):
|
|
raise serializers.ValidationError(_('Old password mismatch.'))
|
|
# Compare new password with the old ones
|
|
if self.instance.check_password(raw_password=password):
|
|
raise serializers.ValidationError(_('Password is already in use'))
|
|
# Validate password
|
|
password_validators.validate_password(password=password)
|
|
except serializers.ValidationError as e:
|
|
raise serializers.ValidationError({'detail': e.detail})
|
|
else:
|
|
if settings.USE_CELERY:
|
|
tasks.send_password_changed_email(
|
|
user_id=self.instance.id,
|
|
country_code=self.context.get('request').country_code)
|
|
else:
|
|
tasks.send_password_changed_email(
|
|
user_id=self.instance.id,
|
|
country_code=self.context.get('request').country_code)
|
|
return attrs
|
|
|
|
def update(self, instance, validated_data):
|
|
"""Override update method"""
|
|
# Update user password from instance
|
|
instance.set_password(validated_data.get('password'))
|
|
instance.save()
|
|
return instance
|
|
|
|
|
|
class ChangeEmailSerializer(serializers.ModelSerializer):
|
|
"""Change user email serializer"""
|
|
|
|
class Meta:
|
|
"""Meta class"""
|
|
model = models.User
|
|
fields = (
|
|
'email',
|
|
)
|
|
|
|
def validate_email(self, value):
|
|
"""Validate email value"""
|
|
if value == self.instance.email:
|
|
raise serializers.ValidationError()
|
|
return value
|
|
|
|
def validate(self, attrs):
|
|
"""Override validate method"""
|
|
email_confirmed = self.instance.email_confirmed
|
|
if not email_confirmed:
|
|
raise serializers.ValidationError()
|
|
return attrs
|
|
|
|
def update(self, instance, validated_data):
|
|
"""
|
|
Override update method
|
|
"""
|
|
instance.email = validated_data.get('email')
|
|
instance.email_confirmed = False
|
|
instance.save()
|
|
# Send verification link on user email for change email address
|
|
if settings.USE_CELERY:
|
|
tasks.confirm_new_email_address.delay(instance.id)
|
|
else:
|
|
tasks.confirm_new_email_address(instance.id)
|
|
return instance
|
|
|
|
|
|
# Firebase Cloud Messaging serializers
|
|
class FCMDeviceSerializer(serializers.ModelSerializer):
|
|
"""FCM Device model serializer"""
|
|
|
|
class Meta:
|
|
model = FCMDevice
|
|
fields = ('id', 'name', 'registration_id', 'device_id',
|
|
'active', 'date_created', 'type')
|
|
read_only_fields = ('id', 'date_created',)
|
|
extra_kwargs = {'active': {'default': True}}
|
|
|
|
def validate(self, attrs):
|
|
regid = attrs.get('registration_id')
|
|
dtype = attrs.get('type')
|
|
if regid and dtype and self.Meta.model.objects.filter(
|
|
registration_id=regid).exclude(type=dtype).count():
|
|
raise exceptions.ValidationError(
|
|
{'registration_id': 'This field must be unique.'})
|
|
return attrs
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(FCMDeviceSerializer, self).__init__(*args, **kwargs)
|
|
self.fields['type'].help_text = (
|
|
'Should be one of ["%s"]' %
|
|
'", "'.join([i for i in self.fields['type'].choices]))
|
|
|
|
def create(self, validated_data):
|
|
user = self.context['request'].user
|
|
if not user.is_anonymous:
|
|
validated_data['user'] = user
|
|
device = FCMDevice.objects.create(**validated_data)
|
|
return device
|
|
|
|
def update(self, instance, validated_data):
|
|
|
|
user = self.context['request'].user
|
|
if not user.is_anonymous:
|
|
instance.user = user
|
|
instance.save()
|
|
else:
|
|
instance.user = None
|
|
instance.save()
|
|
return instance
|