206 lines
9 KiB
Python
206 lines
9 KiB
Python
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)
|