"""Account view """ import threading from notifications.utils import remove_fcm_token # django imports from rest_framework.viewsets import GenericViewSet, mixins from datetime import datetime, timedelta from rest_framework import viewsets, status, views from rest_framework.decorators import action import logging from django.utils import timezone import jwt from django.contrib.auth import logout from django.contrib.auth import authenticate, login from rest_framework.permissions import IsAuthenticated from templated_email import send_templated_mail import google.oauth2.credentials import google.auth.transport.requests from rest_framework import status import requests from rest_framework.response import Response from rest_framework import mixins from django.conf import settings # local imports from guardian.models import Guardian from junior.models import Junior, JuniorPoints from guardian.utils import upload_image_to_alibaba from account.models import (UserDeviceDetails, UserPhoneOtp, UserEmailOtp, DefaultTaskImages, UserNotification, ForceUpdate) from django.contrib.auth.models import User from .serializers import (SuperUserSerializer, GuardianSerializer, JuniorSerializer, EmailVerificationSerializer, ForgotPasswordSerializer, ResetPasswordSerializer, ChangePasswordSerializer, GoogleLoginSerializer, UpdateGuardianImageSerializer, UpdateJuniorProfileImageSerializer, DefaultTaskImagesSerializer, DefaultTaskImagesDetailsSerializer, UserDeleteSerializer, UserNotificationSerializer, UpdateUserNotificationSerializer, UserPhoneOtpSerializer, AdminLoginSerializer, ForceUpdateSerializer) from rest_framework_simplejwt.tokens import RefreshToken from base.messages import ERROR_CODE, SUCCESS_CODE from base.constants import NUMBER, ZOD, JUN, GRD, USER_TYPE_FLAG from guardian.tasks import generate_otp from account.utils import (send_otp_email, send_support_email, custom_response, custom_error_response, generate_code, OTP_EXPIRY, user_device_details, send_all_email) from junior.serializers import JuniorProfileSerializer from guardian.serializers import GuardianProfileSerializer class GoogleLoginMixin(object): """google login mixin""" @staticmethod def google_login(request): """google login function""" access_token = request.data.get('access_token') user_type = request.META.get('HTTP_USER_TYPE') device_id = request.META.get('HTTP_DEVICE_ID') if not access_token: return Response({'error': 'Access token is required.'}, status=status.HTTP_400_BAD_REQUEST) try: # Validate the access token and obtain the user's email and name credentials = google.oauth2.credentials.Credentials.from_authorized_user_info( info={ 'access_token': access_token, 'token_uri': 'https://oauth2.googleapis.com/token', 'client_id': settings.GOOGLE_CLIENT_ID, 'client_secret': settings.GOOGLE_CLIENT_SECRET, 'refresh_token': None, } ) user_info_endpoint = f'https://www.googleapis.com/oauth2/v3/userinfo?access_token={access_token}' headers = {'Authorization': f'Bearer {credentials.token}'} response = requests.get(user_info_endpoint, headers=headers) response.raise_for_status() user_info = response.json() email = user_info['email'] first_name = user_info['given_name'] last_name = user_info['family_name'] if 'family_name' in user_info and user_info['family_name'] else user_info['given_name'] profile_picture = user_info['picture'] except Exception as e: return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) # Check if the user exists in your database or create a new user # ... user_data = User.objects.filter(email__iexact=email) if user_data.exists(): if str(user_type) == '1': junior_query = Junior.objects.filter(auth=user_data.last()).last() if not junior_query: return custom_error_response( ERROR_CODE["2071"], response_status=status.HTTP_400_BAD_REQUEST ) if not junior_query.is_active: return custom_error_response( ERROR_CODE["2075"], response_status=status.HTTP_404_NOT_FOUND ) serializer = JuniorSerializer(junior_query) elif str(user_type) == '2': guardian_query = Guardian.objects.filter(user=user_data.last()).last() if not guardian_query: return custom_error_response( ERROR_CODE["2070"], response_status=status.HTTP_400_BAD_REQUEST ) if not guardian_query.is_active: return custom_error_response( ERROR_CODE["2075"], response_status=status.HTTP_404_NOT_FOUND ) serializer = GuardianSerializer(guardian_query) else: return custom_error_response( ERROR_CODE["2069"], response_status=status.HTTP_400_BAD_REQUEST ) device_detail, created = UserDeviceDetails.objects.get_or_create(user=user_data.last()) if device_detail: device_detail.device_id = device_id device_detail.save() return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) else: user_obj = User.objects.create(username=email, email=email, first_name=first_name, last_name=last_name) if str(user_type) == '1': junior_query = Junior.objects.create(auth=user_obj, is_verified=True, is_active=True, image=profile_picture, signup_method='2', junior_code=generate_code(JUN, user_obj.id), referral_code=generate_code(ZOD, user_obj.id) ) serializer = JuniorSerializer(junior_query, context={'sign_up': True}) position = Junior.objects.all().count() JuniorPoints.objects.create(junior=junior_query, position=position) elif str(user_type) == '2': guardian_query = Guardian.objects.create(user=user_obj, is_verified=True, is_active=True, image=profile_picture,signup_method='2', guardian_code=generate_code(GRD, user_obj.id), referral_code=generate_code(ZOD, user_obj.id) ) serializer = GuardianSerializer(guardian_query, context={'sign_up': True}) else: user_obj.delete() return custom_error_response( ERROR_CODE["2069"], response_status=status.HTTP_400_BAD_REQUEST ) device_detail, created = UserDeviceDetails.objects.get_or_create(user=user_obj) if device_detail: device_detail.device_id = device_id device_detail.save() # Return a JSON response with the user's email and name return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) class GoogleLoginViewSet(GoogleLoginMixin, viewsets.GenericViewSet): """Google login viewset""" serializer_class = GoogleLoginSerializer def create(self, request): """Payload { "access_token", "user_type": "1" }""" serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) return self.google_login(request) class SigninWithApple(views.APIView): """This API is for sign in with Apple for app. Payload { "access_token", "user_type": "1" }""" def post(self, request): token = request.data.get("access_token") user_type = request.META.get('HTTP_USER_TYPE') device_id = request.META.get('HTTP_DEVICE_ID') try: decoded_data = jwt.decode(token, options={"verify_signature": False}) user_data = {"email": decoded_data.get('email'), "username": decoded_data.get('email'), "is_active": True} if decoded_data.get("email"): try: user = User.objects.get(email__iexact=decoded_data.get("email")) if str(user_type) == '1': junior_data = Junior.objects.filter(auth=user).last() if not junior_data: return custom_error_response( ERROR_CODE["2071"], response_status=status.HTTP_400_BAD_REQUEST ) serializer = JuniorSerializer(junior_data) elif str(user_type) == '2': guardian_data = Guardian.objects.filter(user=user).last() if not guardian_data: return custom_error_response( ERROR_CODE["2070"], response_status=status.HTTP_400_BAD_REQUEST ) serializer = GuardianSerializer(guardian_data) else: return custom_error_response( ERROR_CODE["2069"], response_status=status.HTTP_400_BAD_REQUEST ) device_detail, created = UserDeviceDetails.objects.get_or_create(user=user) if device_detail: device_detail.device_id = device_id device_detail.save() return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) except User.DoesNotExist: user = User.objects.create(**user_data) if str(user_type) == '1': junior_query = Junior.objects.create(auth=user, is_verified=True, is_active=True, signup_method='3', junior_code=generate_code(JUN, user.id), referral_code=generate_code(ZOD, user.id)) serializer = JuniorSerializer(junior_query, context={'sign_up': True}) position = Junior.objects.all().count() JuniorPoints.objects.create(junior=junior_query, position=position) elif str(user_type) == '2': guardian_query = Guardian.objects.create(user=user, is_verified=True, is_active=True, signup_method='3', guardian_code=generate_code(GRD, user.id), referral_code=generate_code(ZOD, user.id)) serializer = GuardianSerializer(guardian_query, context={'sign_up': True}) else: user.delete() return custom_error_response( ERROR_CODE["2069"], response_status=status.HTTP_400_BAD_REQUEST ) device_detail, created = UserDeviceDetails.objects.get_or_create(user=user) if device_detail: device_detail.device_id = device_id device_detail.save() return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) except Exception as e: logging.error(e) return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) class UpdateProfileImage(views.APIView): """Update profile image""" permission_classes = [IsAuthenticated] def put(self, request, format=None): try: image = request.data['image'] if image and image.size == NUMBER['zero']: return custom_error_response(ERROR_CODE['2035'], response_status=status.HTTP_400_BAD_REQUEST) filename = f"images/{image.name}" image_url = upload_image_to_alibaba(image, filename) image_data = image_url if str(request.data['user_type']) == '1': junior_query = Junior.objects.filter(auth=request.user).last() serializer = UpdateJuniorProfileImageSerializer(junior_query, data={'image':image_data}, partial=True) elif str(request.data['user_type']) == '2': guardian_query = Guardian.objects.filter(user=request.user).last() serializer = UpdateGuardianImageSerializer(guardian_query, data={'image':image_data}, partial=True) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3017'], serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) except Exception as e: logging.error(e) return custom_error_response(ERROR_CODE['2036'],response_status=status.HTTP_400_BAD_REQUEST) class ChangePasswordAPIView(views.APIView): """ change password" """ serializer_class = ChangePasswordSerializer permission_classes = [IsAuthenticated] def post(self, request): """ POST request to change current login user password Payload { "current_password":"Demo@123", "new_password":"Demo@123" } """ serializer = ChangePasswordSerializer( context=request.user, data=request.data ) if serializer.is_valid(): serializer.save() return custom_response( SUCCESS_CODE['3007'], response_status=status.HTTP_200_OK ) return custom_error_response( serializer.errors, response_status=status.HTTP_400_BAD_REQUEST ) class ResetPasswordAPIView(views.APIView): """Reset password Payload { "verification_code":"373770", "password":"Demo@1323" }""" def post(self, request): serializer = ResetPasswordSerializer(data=request.data) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3006'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class ForgotPasswordAPIView(views.APIView): """ Forgot password """ serializer_class = ForgotPasswordSerializer def post(self, request): """ Payload { "email": "abc@yopmail.com" } """ serializer = self.serializer_class(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data['email'] # generate otp verification_code = generate_otp() # Send the verification code to the user's email send_all_email.delay( 'email_reset_verification.email', email, verification_code ) expiry = timezone.now() + timezone.timedelta(days=1) user_data, created = UserEmailOtp.objects.get_or_create( email=email ) if created: user_data.expired_at = expiry user_data.save() if user_data: user_data.otp = verification_code user_data.expired_at = expiry user_data.save() return custom_response( SUCCESS_CODE['3015'], response_status=status.HTTP_200_OK ) class SendPhoneOtp(viewsets.ModelViewSet): """Send otp on phone""" serializer_class = UserPhoneOtpSerializer def create(self, request, *args, **kwargs): otp = generate_otp() phone_number = self.request.data['phone'] if phone_number.isdigit() and len(phone_number) == 10: phone_otp, created = UserPhoneOtp.objects.get_or_create(country_code=self.request.data['country_code'], phone=self.request.data['phone']) if phone_otp: phone_otp.otp = otp phone_otp.save() return custom_response(None, {'phone_otp':otp}, response_status=status.HTTP_200_OK) return custom_error_response(ERROR_CODE['2020'], response_status=status.HTTP_400_BAD_REQUEST) class UserPhoneVerification(viewsets.ModelViewSet): """Send otp on phone""" serializer_class = UserPhoneOtpSerializer def list(self, request, *args, **kwargs): try: phone_data = UserPhoneOtp.objects.filter(phone=self.request.GET.get('phone'), otp=self.request.GET.get('otp')).last() if phone_data: phone_data.is_verified = True phone_data.save() return custom_response(SUCCESS_CODE['3012'], response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) except Exception: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) class UserLogin(viewsets.ViewSet): """User login""" @action(methods=['post'], detail=False) def login(self, request): username = request.data.get('username') password = request.data.get('password') user_type = request.META.get('HTTP_USER_TYPE') device_id = request.META.get('HTTP_DEVICE_ID') user = authenticate(request, username=username, password=password) try: if user is not None: login(request, user) if str(user_type) == USER_TYPE_FLAG["TWO"]: guardian_data = Guardian.objects.filter(user__username=username).last() if guardian_data: if guardian_data.is_verified: serializer = GuardianSerializer( guardian_data, context={'user_type': user_type} ).data else: return custom_error_response( ERROR_CODE["2070"], response_status=status.HTTP_401_UNAUTHORIZED ) elif str(user_type) == USER_TYPE_FLAG["FIRST"]: junior_data = Junior.objects.filter(auth__username=username).last() if junior_data: if junior_data.is_verified: serializer = JuniorSerializer( junior_data, context={'user_type': user_type} ).data else: return custom_error_response( ERROR_CODE["2071"], response_status=status.HTTP_401_UNAUTHORIZED ) else: return custom_error_response( ERROR_CODE["2069"], response_status=status.HTTP_401_UNAUTHORIZED ) # storing device id in using celery task so the time would be reduced # user_device_details.delay(user.id, device_id) device_details, created = UserDeviceDetails.objects.get_or_create(user=user) if device_details: device_details.device_id = device_id device_details.save() return custom_response(SUCCESS_CODE['3003'], serializer, response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2002"], response_status=status.HTTP_401_UNAUTHORIZED) except Exception as e: logging.error(e) email_verified = UserEmailOtp.objects.filter(email=username).last() refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) refresh_token = str(refresh) data = { "auth_token":access_token, "refresh_token":refresh_token, "is_profile_complete": False, "user_type": user_type, } is_verified = False if email_verified: is_verified = email_verified.is_verified if not is_verified: otp = generate_otp() email_verified.otp = otp email_verified.save() data.update({"email_otp":otp}) send_otp_email.delay(username, otp) return custom_response( ERROR_CODE['2024'], {"email_otp": otp, "is_email_verified": is_verified}, response_status=status.HTTP_200_OK ) data.update({"is_email_verified": is_verified}) return custom_response( SUCCESS_CODE['3003'], data, response_status=status.HTTP_200_OK ) @action(methods=['post'], detail=False) def admin_login(self, request): email = request.data.get('email') password = request.data.get('password') user = User.objects.filter(email__iexact=email, is_superuser=True ).only('id', 'first_name', 'last_name', 'email', 'username', 'is_active', 'is_superuser').first() if not user or not user.check_password(password): return custom_error_response(ERROR_CODE["2002"], response_status=status.HTTP_400_BAD_REQUEST) serializer = SuperUserSerializer(user) return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) class AdminLoginViewSet(viewsets.GenericViewSet): """ admin login api """ serializer_class = AdminLoginSerializer @action(methods=['post'], url_name='login', url_path='login', detail=False) def admin_login(self, request, *args, **kwargs): """ :param request: :return: """ serializer = self.serializer_class(data=request.data) serializer.is_valid(raise_exception=True) user = serializer.save() refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) refresh_token = str(refresh) data = {"auth_token": access_token, "refresh_token": refresh_token, "username": user.username, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "is_active": user.is_active, "user_type": '3', "is_superuser": user.is_superuser} return custom_response(None, data) class UserEmailVerification(viewsets.ModelViewSet): """User Email verification Payload { "email":"ramu@yopmail.com", "otp":"361123" }""" serializer_class = EmailVerificationSerializer http_method_names = ('post',) def create(self, request, *args, **kwargs): try: user_obj = User.objects.filter(username=self.request.data.get('email')).last() email_data = UserEmailOtp.objects.filter(email=self.request.data.get('email'), otp=self.request.data.get('otp')).last() if email_data: input_datetime_str = str(email_data.expired_at) input_format = "%Y-%m-%d %H:%M:%S.%f%z" output_format = "%Y-%m-%d %H:%M:%S.%f" input_datetime = datetime.strptime(input_datetime_str, input_format) output_datetime_str = input_datetime.strftime(output_format) format_str = "%Y-%m-%d %H:%M:%S.%f" datetime_obj = datetime.strptime(output_datetime_str, format_str) if datetime.today() > datetime_obj: return custom_error_response(ERROR_CODE["2029"], response_status=status.HTTP_400_BAD_REQUEST) email_data.is_verified = True email_data.save() if email_data.user_type == '1': junior_data = Junior.objects.filter(auth__email=self.request.data.get('email')).last() if junior_data: junior_data.is_verified = True junior_data.save() else: guardian_data = Guardian.objects.filter(user__email=self.request.data.get('email')).last() if guardian_data: guardian_data.is_verified = True guardian_data.save() refresh = RefreshToken.for_user(user_obj) access_token = str(refresh.access_token) refresh_token = str(refresh) return custom_response(SUCCESS_CODE['3011'], {"auth_token":access_token, "refresh_token":refresh_token}, response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) except Exception as e: logging.error(e) return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) class ReSendEmailOtp(viewsets.ModelViewSet): """Send otp on phone""" serializer_class = EmailVerificationSerializer http_method_names = ('post',) def create(self, request, *args, **kwargs): """Param {"email":"ashok@yopmail.com"} """ otp = generate_otp() if User.objects.filter(email=request.data['email']): expiry = timezone.now() + timezone.timedelta(days=1) email_data, created = UserEmailOtp.objects.get_or_create(email=request.data['email']) if created: email_data.expired_at = expiry email_data.save() if email_data: email_data.otp = otp email_data.expired_at = expiry email_data.save() send_otp_email.delay(request.data['email'], otp) return custom_response(SUCCESS_CODE['3016'], response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2023"], response_status=status.HTTP_400_BAD_REQUEST) class ProfileAPIViewSet(viewsets.ModelViewSet): """Profile viewset""" serializer_class = JuniorProfileSerializer permission_classes = [IsAuthenticated] http_method_names = ('get',) def list(self, request, *args, **kwargs): """profile view Params user_type""" user_type = request.META.get('HTTP_USER_TYPE') if str(user_type) == '1': junior_data = Junior.objects.filter(auth=self.request.user).last() if junior_data: serializer = JuniorProfileSerializer(junior_data) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) elif str(user_type) == '2': guardian_data = Guardian.objects.filter(user=self.request.user).last() if guardian_data: serializer = GuardianProfileSerializer(guardian_data) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(None, response_status=status.HTTP_400_BAD_REQUEST) class UploadImageAPIViewSet(viewsets.ModelViewSet): """upload task image""" serializer_class = DefaultTaskImagesSerializer http_method_names = ('post',) def create(self, request, *args, **kwargs): """upload images""" image_data = request.data['image_url'] filename = f"default_task_images/{image_data.name}" if image_data.size == NUMBER['zero']: return custom_error_response(ERROR_CODE['2035'], response_status=status.HTTP_400_BAD_REQUEST) image = upload_image_to_alibaba(image_data, filename) image_data = image request.data['image_url'] = image_data serializer = DefaultTaskImagesSerializer(data=request.data) if serializer.is_valid(): serializer.save() return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.error, response_status=status.HTTP_400_BAD_REQUEST) class DefaultImageAPIViewSet(viewsets.ModelViewSet): """Profile viewset""" serializer_class = DefaultTaskImagesDetailsSerializer permission_classes = [IsAuthenticated] http_method_names = ('get',) def list(self, request, *args, **kwargs): """profile view""" queryset = DefaultTaskImages.objects.all() serializer = DefaultTaskImagesSerializer(queryset, many=True) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) class DeleteUserProfileAPIViewSet(viewsets.GenericViewSet): """ Delete user API view set {"user_type":1, "signup_method":"1", "password":"Demo@123"} signup_method 1 for manual 2 for google login 3 for apple login""" @action(detail=False, methods=['POST'], url_path='user-account',serializer_class=UserDeleteSerializer, permission_classes=[IsAuthenticated]) def account(self, request): user_type = str(request.data['user_type']) password = request.data.get('password') signup_method = str(request.data.get('signup_method')) serializer = self.get_serializer(data=request.data, context={'request': request, 'user': request.user, 'user_type': user_type, 'password': password, 'signup_method':signup_method}) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3005'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class UserNotificationAPIViewSet(viewsets.ModelViewSet): """notification viewset""" serializer_class = UserNotificationSerializer permission_classes = [IsAuthenticated] http_method_names = ('get',) def list(self, request, *args, **kwargs): """notification view""" queryset = UserNotification.objects.filter(user=request.user) serializer = UserNotificationSerializer(queryset, many=True) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) class UpdateUserNotificationAPIViewSet(viewsets.ModelViewSet): """Update notification viewset""" serializer_class = UpdateUserNotificationSerializer permission_classes = [IsAuthenticated] http_method_names = ('post',) def create(self, request, *args, **kwargs): """Payload {"email_notification": false, "sms_notification": false, "push_notification": false} """ serializer = UpdateUserNotificationSerializer(data=request.data, context=request.user) if serializer.is_valid(): serializer.save() return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.error, response_status=status.HTTP_400_BAD_REQUEST) class SendSupportEmail(views.APIView): """support email api payload name email message """ permission_classes = (IsAuthenticated,) def post(self, request): name = request.data.get('name') sender = request.data.get('email') message = request.data.get('message') if name and sender and message: try: send_support_email(name, sender, message) return custom_response(SUCCESS_CODE['3019'], response_status=status.HTTP_200_OK) except Exception as e: return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) else: return custom_error_response(ERROR_CODE['2033'], response_status=status.HTTP_400_BAD_REQUEST) class LogoutAPIView(views.APIView): """Log out API""" permission_classes = (IsAuthenticated,) def post(self, request): remove_fcm_token( request.auth.payload['user_id'], request.META['HTTP_AUTHORIZATION'].split(" ")[1], request.data.get('registration_id', "")) logout(request) request.session.flush() return custom_response(SUCCESS_CODE['3020'], response_status=status.HTTP_200_OK) class AccessTokenAPIView(views.APIView): """generate access token API""" def post(self, request): # Assuming you have a refresh_token string refresh_token = request.data['refresh_token'] # Create a RefreshToken instance from the refresh token string refresh = RefreshToken(refresh_token) # Generate a new access token access_token = str(refresh.access_token) data = {"auth_token": access_token} return custom_response(None, data, response_status=status.HTTP_200_OK) class ForceUpdateViewSet(GenericViewSet, mixins.CreateModelMixin): """FAQ view set""" serializer_class = ForceUpdateSerializer http_method_names = ['post'] def create(self, request, *args, **kwargs): """ faq create api method :param request: :param args: version, device type :param kwargs: :return: success message """ if ForceUpdate.objects.all().count() >= 4: return custom_error_response(ERROR_CODE['2080'], response_status=status.HTTP_400_BAD_REQUEST) obj_data = [ForceUpdate(**item) for item in request.data] try: ForceUpdate.objects.bulk_create(obj_data) return custom_response(SUCCESS_CODE["3046"], response_status=status.HTTP_200_OK) except Exception as e: return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST)