import logging import traceback from contextlib import suppress from secrets import compare_digest from django.contrib.auth.password_validation import validate_password from django.contrib.auth.tokens import PasswordResetTokenGenerator from django.core.exceptions import ValidationError from django.utils.decorators import method_decorator from django.utils.http import urlsafe_base64_decode from django.utils.translation import gettext_lazy as _ from django_ratelimit.decorators import ratelimit from drf_spectacular.utils import extend_schema_view from rest_framework import mixins, status from rest_framework.decorators import action from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.viewsets import GenericViewSet from rest_framework_simplejwt.tokens import RefreshToken from evibes.settings import DEBUG from vibes_auth.docs.drf.viewsets import USER_SCHEMA from vibes_auth.models import User from vibes_auth.serializers import ( MergeRecentlyViewedSerializer, UserSerializer, ) from vibes_auth.utils.emailing import send_reset_password_email_task logger = logging.getLogger("evibes") @extend_schema_view(**USER_SCHEMA) class UserViewSet( mixins.CreateModelMixin, mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.DestroyModelMixin, GenericViewSet, ): """ User view set implementation using Django REST framework. Provides a set of actions that manage user-related data such as creation, retrieval, updates, deletion, and custom actions including password reset, avatar upload, account activation, and recently viewed items merging. This class extends the mixins and GenericViewSet for robust API handling. Attributes: serializer_class (class): Serializer class used to serialize the user data. queryset (QuerySet): Queryset that fetches active user objects. permission_classes (list): List of permissions applied to certain actions. Methods: reset_password(request): Sends a reset password email to a user based on the provided email. upload_avatar(request, **_kwargs): Allows an authenticated user to upload an avatar for their profile. confirm_password_reset(request, *_args, **_kwargs): Confirms the password reset and updates the new password for a user. create(request, *args, **kwargs): Creates a new user instance using the provided payload. activate(request): Activates a user account based on their unique activation link. merge_recently_viewed(request, **_kwargs): Merges a list of recently viewed products into the user's profile. retrieve(request, pk=None, *args, **kwargs): Retrieves user details for the given user ID. update(request, pk=None, *args, **kwargs): Updates user details for the given user ID. Raises: DoesNotExist: Raised when a user is not found in specific methods working with database queries. ValidationError: Raised for invalid data during serialization or password validation. OverflowError, TypeError: Raised for decoding or type conversion issues. """ serializer_class = UserSerializer queryset = User.objects.filter(is_active=True) permission_classes = [AllowAny] @action(detail=False, methods=["post"]) @method_decorator(ratelimit(key="ip", rate="2/h" if not DEBUG else "888/h")) def reset_password(self, request): user = None with suppress(User.DoesNotExist): user = User.objects.get(email=request.data.get("email")) if user: send_reset_password_email_task.delay(user_pk=user.uuid) return Response(status=status.HTTP_200_OK) @action(detail=True, methods=["put"], permission_classes=[IsAuthenticated]) @method_decorator(ratelimit(key="ip", rate="2/h" if not DEBUG else "888/h")) def upload_avatar(self, request, **_kwargs): user = self.get_object() if request.user != user: return Response(status=status.HTTP_403_FORBIDDEN) if "avatar" in request.FILES: user.avatar = request.FILES["avatar"] user.save() return Response(status=status.HTTP_200_OK, data=self.serializer_class(user).data) return Response(status=status.HTTP_400_BAD_REQUEST) @action(detail=False, methods=["post"]) @method_decorator(ratelimit(key="ip", rate="2/h" if not DEBUG else "888/h")) def confirm_password_reset(self, request, *_args, **_kwargs): try: if not compare_digest(request.data.get("password"), request.data.get("confirm_password")): return Response( {"error": _("passwords do not match")}, status=status.HTTP_400_BAD_REQUEST, ) uuid = urlsafe_base64_decode(request.data.get("uidb_64")).decode() user = User.objects.get(pk=uuid) validate_password(password=request.data.get("password"), user=user) password_reset_token = PasswordResetTokenGenerator() if not password_reset_token.check_token(user, request.data.get("token")): return Response({"error": _("token is invalid!")}, status=status.HTTP_400_BAD_REQUEST) user.set_password(request.data.get("password")) user.save() return Response({"message": _("password reset successfully")}, status=status.HTTP_200_OK) except (TypeError, ValueError, OverflowError, ValidationError, User.DoesNotExist) as e: data = {"error": str(e)} if DEBUG: data["detail"] = str(traceback.format_exc()) data["received"] = str(request.data) return Response(data, status=status.HTTP_400_BAD_REQUEST) @method_decorator(ratelimit(key="ip", rate="3/h" if not DEBUG else "888/h")) def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) user = serializer.save() user.save() headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) @action(detail=False, methods=["post"]) @method_decorator(ratelimit(key="ip", rate="2/h" if not DEBUG else "888/h")) def activate(self, request): detail = "" activation_error = None try: uuid = urlsafe_base64_decode(request.data.get("uidb_64")).decode() user = User.objects.nocache().get(pk=uuid) if not user.check_token(urlsafe_base64_decode(request.data.get("token")).decode()): return Response( {"error": _("activation link is invalid!")}, status=status.HTTP_400_BAD_REQUEST, ) if user.is_active: return Response( {"error": _("account already activated!")}, status=status.HTTP_400_BAD_REQUEST, ) user.is_active = True user.is_verified = True user.save() except (TypeError, ValueError, OverflowError, User.DoesNotExist) as activation_error: user = None activation_error = activation_error detail = str(traceback.format_exc()) if user is None: if DEBUG: raise Exception from activation_error return Response( {"error": _("activation link is invalid!"), "detail": detail}, status=status.HTTP_400_BAD_REQUEST, ) else: tokens = RefreshToken.for_user(user) response_data = self.serializer_class(user).data response_data["refresh"] = str(tokens) response_data["access"] = str(tokens.access_token) return Response(response_data, status=status.HTTP_200_OK) @action(detail=True, methods=["put"], permission_classes=[IsAuthenticated]) def merge_recently_viewed(self, request, **_kwargs): user = self.get_object() if request.user != user: return Response(status=status.HTTP_403_FORBIDDEN) serializer = MergeRecentlyViewedSerializer(data=request.data, context=self.get_serializer_context()) serializer.is_valid(raise_exception=True) for product_uuid in serializer.validated_data["product_uuids"]: user.add_to_recently_viewed(product_uuid) return Response(status=status.HTTP_202_ACCEPTED, data=self.serializer_class(user).data) def retrieve(self, request, pk=None, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) return Response(serializer.data) def update(self, request, pk=None, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) instance = serializer.update(instance=self.get_object(), validated_data=request.data) return Response(self.get_serializer(instance).data)