File: //home/arjun/projects/unlimited-leads/Unlimited-Leads-Be/authorization/auth_views/verify_account.py
from datetime import timedelta
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_simplejwt.tokens import RefreshToken
from django.utils import timezone
from ..models import VerificationMailRequest, LoginUserDevice
from ..serializers import VerifyAccountSerializer
from utils.utils import get_first_error
class VerifyAccountView(APIView):
@swagger_auto_schema(
operation_summary="Verify OTP sent to email",
tags=["Auth"],
request_body=VerifyAccountSerializer,
responses={
status.HTTP_200_OK: openapi.Response(
description="OTP verified successful",
schema=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"data": openapi.Schema(type=openapi.TYPE_OBJECT),
"message": openapi.Schema(type=openapi.TYPE_STRING),
"success": openapi.Schema(type=openapi.TYPE_BOOLEAN),
"statusCode": openapi.Schema(type=openapi.TYPE_INTEGER),
},
),
),
status.HTTP_400_BAD_REQUEST: openapi.Response(
description="Bad request, validation error.",
schema=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"errors": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(type=openapi.TYPE_STRING),
),
"message": openapi.Schema(type=openapi.TYPE_STRING),
"success": openapi.Schema(type=openapi.TYPE_BOOLEAN),
"statusCode": openapi.Schema(type=openapi.TYPE_INTEGER),
},
),
),
},
)
def post(self, request):
serializer = VerifyAccountSerializer(data=request.data)
if not serializer.is_valid():
return Response(
{
"success": False,
"message": get_first_error(
serializer.errors, "Verification failed"
),
"statusCode": status.HTTP_400_BAD_REQUEST,
},
status=status.HTTP_400_BAD_REQUEST,
)
user = serializer.validated_data["email"]
device_type = serializer.validated_data["device_type"]
is_remember = serializer.validated_data["is_remember"]
last_login = timezone.now()
verification = VerificationMailRequest.objects.get(user=user)
verification.otp = None
verification.save()
user.is_verified = True
user.is_active = True
user.last_login = last_login
user.save()
try:
device = LoginUserDevice.objects.get(user=user, device_type=device_type)
device.login_at = last_login
device.is_remember = is_remember
except LoginUserDevice.DoesNotExist:
device = LoginUserDevice.objects.create(
user=user,
device_type=device_type,
login_at=last_login,
is_remember=is_remember,
)
user.last_login = last_login
user.save()
refresh = RefreshToken.for_user(user)
if device_type == "browser" and is_remember:
device.expiration_period = 30
refresh.set_exp(lifetime=timedelta(days=device.expiration_period))
elif device_type == "browser":
device.expiration_period = 1
device.save()
return Response(
{
"data": {
"id": user.id,
"token": {
"access": str(refresh.access_token),
"refresh": str(refresh),
},
},
"success": True,
"message": "Login successful",
"statusCode": status.HTTP_200_OK,
},
status=status.HTTP_200_OK,
)