210 lines
7.9 KiB
Python
210 lines
7.9 KiB
Python
import uuid
|
|
from typing import Type
|
|
|
|
from core.services import UserService, SmsService
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django_core import exceptions
|
|
from drf_spectacular.utils import extend_schema
|
|
from rest_framework import status, throttling, request
|
|
from rest_framework.response import Response
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.viewsets import GenericViewSet
|
|
from django_core.mixins import BaseViewSetMixin
|
|
from rest_framework.decorators import action
|
|
from ..serializers import (
|
|
RegisterSerializer,
|
|
ConfirmSerializer,
|
|
ResendSerializer,
|
|
ResetPasswordSerializer,
|
|
ResetConfirmationSerializer,
|
|
SetPasswordSerializer,
|
|
UserSerializer,
|
|
UserUpdateSerializer,
|
|
)
|
|
from rest_framework.permissions import AllowAny
|
|
from django.contrib.auth.hashers import make_password
|
|
from drf_spectacular.utils import OpenApiResponse
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from ..serializers import ChangePasswordSerializer
|
|
|
|
from .. import models
|
|
|
|
|
|
@extend_schema(tags=["register"])
|
|
class RegisterView(BaseViewSetMixin, GenericViewSet, UserService):
|
|
throttle_classes = [throttling.UserRateThrottle]
|
|
permission_classes = [AllowAny]
|
|
|
|
def get_serializer_class(self):
|
|
match self.action:
|
|
case "register":
|
|
return RegisterSerializer
|
|
case "confirm":
|
|
return ConfirmSerializer
|
|
case "resend":
|
|
return ResendSerializer
|
|
case _:
|
|
return RegisterSerializer
|
|
|
|
@action(methods=["POST"], detail=False, url_path="register")
|
|
def register(self, request):
|
|
ser = self.get_serializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
data = ser.data
|
|
phone = data.get("phone")
|
|
# Create pending user
|
|
self.create_user(phone)
|
|
self.send_confirmation(phone) # Send confirmation code for sms eskiz.uz
|
|
return Response(
|
|
{"detail": _("Sms %(phone)s raqamiga yuborildi") % {"phone": phone}},
|
|
status=status.HTTP_202_ACCEPTED,
|
|
)
|
|
|
|
@extend_schema(summary="Auth confirm.", description="Auth confirm user.")
|
|
@action(methods=["POST"], detail=False, url_path="confirm")
|
|
def confirm(self, request):
|
|
ser = self.get_serializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
data = ser.data
|
|
phone, code = data.get("phone"), data.get("code")
|
|
try:
|
|
if SmsService.check_confirm(phone, code=code):
|
|
token = self.validate_user(get_user_model().objects.filter(phone=phone).first())
|
|
return Response(
|
|
data={
|
|
"detail": _("Tasdiqlash ko'di qabul qilindi"),
|
|
"token": token,
|
|
},
|
|
status=status.HTTP_202_ACCEPTED,
|
|
)
|
|
except exceptions.SmsException as e:
|
|
raise PermissionDenied(e) # Response exception for APIException
|
|
except Exception as e:
|
|
raise PermissionDenied(e) # Api exception for APIException
|
|
|
|
@action(methods=["POST"], detail=False, url_path="resend")
|
|
def resend(self, rq: Type[request.Request]):
|
|
ser = self.get_serializer(data=rq.data)
|
|
ser.is_valid(raise_exception=True)
|
|
phone = ser.data.get("phone")
|
|
self.send_confirmation(phone)
|
|
return Response({"detail": _("Sms %(phone)s raqamiga yuborildi") % {"phone": phone}})
|
|
|
|
|
|
@extend_schema(tags=["reset-password"])
|
|
class ResetPasswordView(BaseViewSetMixin, GenericViewSet, UserService):
|
|
permission_classes = [AllowAny]
|
|
|
|
def get_serializer_class(self):
|
|
match self.action:
|
|
case "reset_password":
|
|
return ResetPasswordSerializer
|
|
case "reset_confirm":
|
|
return ResetConfirmationSerializer
|
|
case "reset_password_set":
|
|
return SetPasswordSerializer
|
|
case _:
|
|
return None
|
|
|
|
@action(methods=["POST"], detail=False, url_path="reset-password")
|
|
def reset_password(self, request):
|
|
ser = self.get_serializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
phone = ser.data.get("phone")
|
|
self.send_confirmation(phone)
|
|
return Response({"detail": _("Sms %(phone)s raqamiga yuborildi") % {"phone": phone}})
|
|
|
|
@action(methods=["POST"], detail=False, url_path="reset-password-confirm")
|
|
def reset_confirm(self, request):
|
|
ser = self.get_serializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
|
|
data = ser.data
|
|
code, phone = data.get("code"), data.get("phone")
|
|
try:
|
|
SmsService.check_confirm(phone, code)
|
|
token = models.ResetToken.objects.create(
|
|
user=get_user_model().objects.filter(phone=phone).first(),
|
|
token=str(uuid.uuid4()),
|
|
)
|
|
return Response(
|
|
data={
|
|
"token": token.token,
|
|
"created_at": token.created_at,
|
|
"updated_at": token.updated_at,
|
|
},
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
except exceptions.SmsException as e:
|
|
raise PermissionDenied(str(e))
|
|
except Exception as e:
|
|
raise PermissionDenied(str(e))
|
|
|
|
@action(methods=["POST"], detail=False, url_path="reset-password-set")
|
|
def reset_password_set(self, request):
|
|
ser = self.get_serializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
data = ser.data
|
|
token = data.get("token")
|
|
password = data.get("password")
|
|
token = models.ResetToken.objects.filter(token=token)
|
|
if not token.exists():
|
|
raise PermissionDenied(_("Invalid token"))
|
|
phone = token.first().user.phone
|
|
token.delete()
|
|
self.change_password(phone, password)
|
|
return Response({"detail": _("password updated")}, status=status.HTTP_200_OK)
|
|
|
|
|
|
@extend_schema(tags=["me"])
|
|
class MeView(BaseViewSetMixin, GenericViewSet, UserService):
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_serializer_class(self):
|
|
match self.action:
|
|
case "me":
|
|
return UserSerializer
|
|
case "user_update":
|
|
return UserUpdateSerializer
|
|
case _:
|
|
return None
|
|
|
|
@action(methods=["GET", "OPTIONS"], detail=False, url_path="me")
|
|
def me(self, request):
|
|
return Response(self.get_serializer(request.user).data)
|
|
|
|
@action(methods=["PATCH", "PUT"], detail=False, url_path="user-update")
|
|
def user_update(self, request):
|
|
ser = self.get_serializer(instance=request.user, data=request.data, partial=True)
|
|
ser.is_valid(raise_exception=True)
|
|
ser.save()
|
|
return Response({"detail": _("Malumotlar yangilandi")})
|
|
|
|
|
|
@extend_schema(tags=["change-password"], description="Parolni o'zgartirish uchun")
|
|
class ChangePasswordView(BaseViewSetMixin, GenericViewSet):
|
|
serializer_class = ChangePasswordSerializer
|
|
permission_classes = (IsAuthenticated,)
|
|
|
|
@extend_schema(
|
|
request=serializer_class,
|
|
responses={200: OpenApiResponse(ChangePasswordSerializer)},
|
|
summary="Change user password.",
|
|
description="Change password of the authenticated user.",
|
|
)
|
|
@action(methods=["POST"], detail=False, url_path="change-password")
|
|
def change_password(self, request, *args, **kwargs):
|
|
user = self.request.user
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
if user.check_password(request.data["old_password"]):
|
|
user.password = make_password(request.data["new_password"])
|
|
user.save()
|
|
return Response(
|
|
data={"detail": "password changed successfully"},
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
raise PermissionDenied(_("invalida password"))
|