from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from rest_framework import serializers User = get_user_model() class PermissionSerializer(serializers.ModelSerializer): """A single Django permission, identified by its 'app_label.codename' key.""" key = serializers.SerializerMethodField() app_label = serializers.CharField(source="content_type.app_label", read_only=True) model = serializers.CharField(source="content_type.model", read_only=True) class Meta: model = Permission fields = ["id", "key", "name", "codename", "app_label", "model"] def get_key(self, obj): return f"{obj.content_type.app_label}.{obj.codename}" class AdminUserSerializer(serializers.ModelSerializer): """List / retrieve / update an account. Admin-facing.""" permissions = serializers.SerializerMethodField() groups = serializers.SerializerMethodField() is_locked = serializers.SerializerMethodField() password = serializers.CharField(write_only=True, required=False, allow_blank=False) # Login is by email; username is auto-set to the email when omitted. username = serializers.CharField(required=False) class Meta: model = User fields = [ "id", "email", "username", "role", "full_name", "phone", "location", "skills", "experience_years", "bio", "employee_id", "department", "is_active", "mfa_enabled", "mfa_method", "last_login", "date_joined", "permissions", "groups", "is_locked", "password", ] read_only_fields = [ "id", "mfa_enabled", "mfa_method", "last_login", "date_joined", "permissions", "groups", "is_locked", # Employee IDs are always generated by the server on create and are # never editable — the client value (if any) is ignored. "employee_id", ] def get_is_locked(self, obj): attempt = getattr(obj, "login_attempt", None) return bool(attempt and attempt.is_locked) def get_permissions(self, obj): return sorted(obj.get_all_permissions()) def get_groups(self, obj): return [{'id': g.id, 'name': g.name} for g in obj.groups.all()] def validate_email(self, value): if not value or not value.strip(): raise serializers.ValidationError("Email address is required.") email_clean = value.strip().lower() qs = User.objects.filter(email__iexact=email_clean) if self.instance: qs = qs.exclude(pk=self.instance.pk) if qs.exists(): raise serializers.ValidationError("A user with this email address already exists.") return email_clean def validate_role(self, value): # Roles are dynamic — any existing Group name is a valid role. if not Group.objects.filter(name=value).exists(): existing = list(Group.objects.values_list("name", flat=True)) raise serializers.ValidationError( f"Unknown role '{value}'. Existing roles: {', '.join(existing)}" ) return value def create(self, validated_data): from django.db import IntegrityError from .models import generate_employee_id password = validated_data.pop("password", None) if not password: raise serializers.ValidationError({"password": "Password is required."}) validated_data.setdefault("username", validated_data["email"]) # employee_id is read-only, so it never arrives from the client — the # server always generates it. Retry once on the (very unlikely) unique # collision to stay robust against any legacy/manual ID clash. for _ in range(5): validated_data["employee_id"] = generate_employee_id() user = User(**validated_data) user.set_password(password) # post_save signals sync MCP + role group try: user.save() return user except IntegrityError as err: err_str = str(err).lower() if "email" in err_str or "username" in err_str or "unique constraint" in err_str: raise serializers.ValidationError( {"email": "A user with this email address already exists."} ) continue raise serializers.ValidationError( {"employee_id": "Could not generate a unique Employee ID. Please try again."} ) def update(self, instance, validated_data): password = validated_data.pop("password", None) for attr, value in validated_data.items(): setattr(instance, attr, value) if password: instance.set_password(password) instance.save() return instance class SetRoleSerializer(serializers.Serializer): role = serializers.CharField() def validate_role(self, value): if not Group.objects.filter(name=value).exists(): raise serializers.ValidationError(f"Unknown role '{value}'.") return value class RoleCreateSerializer(serializers.Serializer): """Create a new role (Django Group). Name is normalised to UPPER_SNAKE.""" name = serializers.CharField(max_length=50) permissions = serializers.ListField( child=serializers.CharField(), required=False, allow_empty=True, default=list ) def validate_name(self, value): name = value.strip().upper().replace(" ", "_") if not name: raise serializers.ValidationError("Role name is required.") if Group.objects.filter(name=name).exists(): raise serializers.ValidationError(f"Role '{name}' already exists.") return name def validate_permissions(self, keys): resolved, invalid = [], [] for key in keys: if "." not in key: invalid.append(key) continue app_label, codename = key.split(".", 1) perm = Permission.objects.filter( content_type__app_label=app_label, codename=codename ).first() (resolved if perm else invalid).append(perm if perm else key) if invalid: raise serializers.ValidationError(f"Unknown permission(s): {', '.join(invalid)}") self.context["resolved_permissions"] = resolved return keys class PermissionKeysSerializer(serializers.Serializer): """Replace the full permission set for a role-group or a single user.""" permissions = serializers.ListField( child=serializers.CharField(), allow_empty=True ) def validate_permissions(self, keys): resolved = [] invalid = [] for key in keys: if "." not in key: invalid.append(key) continue app_label, codename = key.split(".", 1) perm = Permission.objects.filter( content_type__app_label=app_label, codename=codename ).first() if perm: resolved.append(perm) else: invalid.append(key) if invalid: raise serializers.ValidationError( f"Unknown permission(s): {', '.join(invalid)}" ) self.context["resolved_permissions"] = resolved return keys class AdminUserListSerializer(AdminUserSerializer): """User-table (bulk list) view — phone + email masked (DPDP). Detail keeps full.""" phone = serializers.SerializerMethodField() email = serializers.SerializerMethodField() def get_phone(self, obj): from core.masking import mask_phone return mask_phone(getattr(obj, "phone", "")) def get_email(self, obj): from core.masking import mask_email return mask_email(getattr(obj, "email", ""))