mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2026-03-10 20:41:46 +00:00
2
.gitignore
vendored
2
.gitignore
vendored
@ -21,4 +21,4 @@ static/*
|
||||
__pycache__/
|
||||
*.env
|
||||
ve/*
|
||||
|
||||
celerybeat-schedule
|
||||
@ -88,7 +88,9 @@ ERROR_CODE = {
|
||||
"2060": "Task does not exist or not in pending state",
|
||||
"2061": "Please insert image or check the image is valid or not.",
|
||||
# email not null
|
||||
"2062": "Please enter email address"
|
||||
"2062": "Please enter email address",
|
||||
"2063": "Unauthorized access.",
|
||||
"2064": "To change your password first request an OTP and get it verify then change your password."
|
||||
}
|
||||
"""Success message code"""
|
||||
SUCCESS_CODE = {
|
||||
|
||||
Binary file not shown.
@ -52,7 +52,6 @@ def send_notification(notification_type, from_user_id, to_user_id, extra_data):
|
||||
""" used to send the push for the given notification type """
|
||||
(notification_data, from_user, to_user) = get_basic_detail(notification_type, from_user_id, to_user_id)
|
||||
user_notification_type = UserNotification.objects.filter(user=to_user).first()
|
||||
# data = notification_data.data
|
||||
data = notification_data
|
||||
Notification.objects.create(notification_type=notification_type, notification_from=from_user,
|
||||
notification_to=to_user, data=data)
|
||||
|
||||
@ -18,7 +18,7 @@ from notifications.utils import send_notification
|
||||
|
||||
class NotificationViewSet(viewsets.GenericViewSet):
|
||||
""" used to do the notification actions """
|
||||
serializer_class = None
|
||||
serializer_class = RegisterDevice
|
||||
permission_classes = [IsAuthenticated, ]
|
||||
|
||||
@action(methods=['post'], detail=False, url_path='device', url_name='device', serializer_class=RegisterDevice)
|
||||
@ -32,7 +32,7 @@ class NotificationViewSet(viewsets.GenericViewSet):
|
||||
serializer.save()
|
||||
return custom_response(SUCCESS_CODE["3000"])
|
||||
|
||||
@action(methods=['get'], detail=False, url_path='test', url_name='test', serializer_class=None)
|
||||
@action(methods=['get'], detail=False, url_path='test', url_name='test')
|
||||
def send_test_notification(self, request):
|
||||
"""
|
||||
to send test notification
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
"""
|
||||
web_admin auth serializers file
|
||||
"""
|
||||
# python imports
|
||||
from datetime import datetime
|
||||
|
||||
# django imports
|
||||
@ -8,41 +9,47 @@ from rest_framework import serializers
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.conf import settings
|
||||
from django.utils import timezone
|
||||
from rest_framework import status
|
||||
from templated_email import send_templated_mail
|
||||
|
||||
# local imports
|
||||
from account.models import UserEmailOtp
|
||||
from account.utils import custom_error_response
|
||||
from base.constants import USER_TYPE
|
||||
from base.messages import ERROR_CODE
|
||||
from guardian.tasks import generate_otp
|
||||
|
||||
USER = get_user_model()
|
||||
|
||||
|
||||
class AdminForgotPasswordSerializer(serializers.ModelSerializer):
|
||||
email = serializers.EmailField(required=True)
|
||||
class AdminOTPSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
admin forgot password serializer
|
||||
"""
|
||||
email = serializers.EmailField()
|
||||
|
||||
class Meta:
|
||||
"""
|
||||
meta class
|
||||
"""
|
||||
model = USER
|
||||
fields = ('email',)
|
||||
|
||||
def validate(self, attrs):
|
||||
""" used to validate the incoming data """
|
||||
user = USER.objects.filter(email=attrs['email']).first()
|
||||
user = USER.objects.filter(email=attrs.get('email')).first()
|
||||
if not user:
|
||||
raise serializers.ValidationError(ERROR_CODE['2004'])
|
||||
elif not user.is_superuser:
|
||||
raise serializers.ValidationError(ERROR_CODE['2036'])
|
||||
raise serializers.ValidationError(ERROR_CODE['2063'])
|
||||
attrs.update({'user': user})
|
||||
return attrs
|
||||
|
||||
def create(self, validated_data):
|
||||
"""
|
||||
to send otp
|
||||
:return: user_data
|
||||
"""
|
||||
email = validated_data['email']
|
||||
try:
|
||||
USER.objects.get(email=email)
|
||||
except USER.DoesNotExist:
|
||||
return custom_error_response(ERROR_CODE['2004'], response_status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
verification_code = generate_otp()
|
||||
|
||||
# Send the verification code to the user's email
|
||||
@ -60,29 +67,81 @@ class AdminForgotPasswordSerializer(serializers.ModelSerializer):
|
||||
user_data, created = UserEmailOtp.objects.get_or_create(email=email)
|
||||
if created:
|
||||
user_data.expired_at = expiry
|
||||
user_data.user_type = dict(USER_TYPE).get('3')
|
||||
if user_data:
|
||||
user_data.otp = verification_code
|
||||
user_data.expired_at = expiry
|
||||
user_data.user_type = dict(USER_TYPE).get('3')
|
||||
user_data.save()
|
||||
return user_data
|
||||
|
||||
|
||||
class AdminVerifyOTPSerializer(serializers.Serializer):
|
||||
"""
|
||||
admin verify otp serializer
|
||||
"""
|
||||
email = serializers.EmailField()
|
||||
otp = serializers.CharField(max_length=6, min_length=6)
|
||||
|
||||
class Meta:
|
||||
""" meta class """
|
||||
fields = ('otp',)
|
||||
fields = ('email', 'otp',)
|
||||
|
||||
def validate(self, attrs):
|
||||
otp = attrs.pop('otp')
|
||||
"""
|
||||
to validate data
|
||||
:return: validated data
|
||||
"""
|
||||
email = attrs.get('email')
|
||||
otp = attrs.get('otp')
|
||||
|
||||
# fetch email otp object of the user
|
||||
user_opt_details = UserEmailOtp.objects.filter(otp=otp).last()
|
||||
if not user_opt_details:
|
||||
raise serializers.ValidationError(ERROR_CODE['2008'])
|
||||
if user_opt_details.expired_at.replace(tzinfo=None) < datetime.utcnow():
|
||||
user_otp_details = UserEmailOtp.objects.filter(email=email, otp=otp).last()
|
||||
if not user_otp_details:
|
||||
raise serializers.ValidationError(ERROR_CODE['2064'])
|
||||
if user_otp_details.user_type != dict(USER_TYPE).get('3'):
|
||||
raise serializers.ValidationError(ERROR_CODE['2063'])
|
||||
if user_otp_details.expired_at.replace(tzinfo=None) < datetime.utcnow():
|
||||
raise serializers.ValidationError(ERROR_CODE['2029'])
|
||||
user_details = USER.objects.filter(email=user_opt_details.email).last()
|
||||
user_opt_details.delete()
|
||||
if user_details:
|
||||
return attrs
|
||||
user_otp_details.is_verified = True
|
||||
user_otp_details.save()
|
||||
return attrs
|
||||
|
||||
|
||||
class AdminCreatePasswordSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
admin create new password serializer
|
||||
"""
|
||||
email = serializers.EmailField()
|
||||
new_password = serializers.CharField()
|
||||
confirm_password = serializers.CharField()
|
||||
|
||||
class Meta:
|
||||
"""
|
||||
meta class
|
||||
"""
|
||||
model = USER
|
||||
fields = ('email', 'new_password', 'confirm_password')
|
||||
|
||||
def validate(self, attrs):
|
||||
"""
|
||||
to validate data
|
||||
:return: validated data
|
||||
"""
|
||||
email = attrs.get('email')
|
||||
new_password = attrs.get('new_password')
|
||||
confirm_password = attrs.get('confirm_password')
|
||||
|
||||
# matching password
|
||||
if new_password != confirm_password:
|
||||
raise serializers.ValidationError('password do not match')
|
||||
|
||||
user_otp_details = UserEmailOtp.objects.filter(email=email).last()
|
||||
if not user_otp_details:
|
||||
raise serializers.ValidationError(ERROR_CODE['2064'])
|
||||
if user_otp_details.user_type != dict(USER_TYPE).get('3'):
|
||||
raise serializers.ValidationError(ERROR_CODE['2063'])
|
||||
if not user_otp_details.is_verified:
|
||||
raise serializers.ValidationError(ERROR_CODE['2064'])
|
||||
user_otp_details.delete()
|
||||
return attrs
|
||||
|
||||
@ -7,7 +7,7 @@ from rest_framework import routers
|
||||
|
||||
# local imports
|
||||
from web_admin.views.article import ArticleViewSet, DefaultArticleCardImagesViewSet, UserManagementViewSet
|
||||
from web_admin.views.auth import ForgetAndResetPasswordViewSet
|
||||
from web_admin.views.auth import ForgotAndResetPasswordViewSet
|
||||
|
||||
# initiate router
|
||||
router = routers.SimpleRouter()
|
||||
@ -17,7 +17,7 @@ router.register('default-card-images', DefaultArticleCardImagesViewSet, basename
|
||||
router.register('user', UserManagementViewSet, basename='user')
|
||||
|
||||
# forgot and reset password api for admin
|
||||
router.register('admin', ForgetAndResetPasswordViewSet, basename='admin')
|
||||
router.register('admin', ForgotAndResetPasswordViewSet, basename='admin')
|
||||
|
||||
urlpatterns = [
|
||||
path('api/v1/', include(router.urls)),
|
||||
|
||||
@ -110,7 +110,7 @@ class ArticleViewSet(GenericViewSet, mixins.CreateModelMixin, mixins.UpdateModel
|
||||
return custom_error_response(ERROR_CODE["2041"], status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
@action(methods=['get'], url_name='remove_card', url_path='remove_card',
|
||||
detail=True, serializer_class=None)
|
||||
detail=True)
|
||||
def remove_article_card(self, request, *args, **kwargs):
|
||||
"""
|
||||
article card remove (delete) api method
|
||||
@ -126,7 +126,7 @@ class ArticleViewSet(GenericViewSet, mixins.CreateModelMixin, mixins.UpdateModel
|
||||
return custom_error_response(ERROR_CODE["2042"], response_status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
@action(methods=['get'], url_name='remove_survey', url_path='remove_survey',
|
||||
detail=True, serializer_class=None)
|
||||
detail=True)
|
||||
def remove_article_survey(self, request, *args, **kwargs):
|
||||
"""
|
||||
article survey remove (delete) api method
|
||||
@ -170,7 +170,7 @@ class DefaultArticleCardImagesViewSet(GenericViewSet, mixins.CreateModelMixin, m
|
||||
api to upload and list default article card images
|
||||
"""
|
||||
serializer_class = DefaultArticleCardImageSerializer
|
||||
# permission_classes = [IsAuthenticated, AdminPermission]
|
||||
permission_classes = [IsAuthenticated, AdminPermission]
|
||||
queryset = DefaultArticleCardImage.objects.all()
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
|
||||
@ -2,26 +2,32 @@
|
||||
web_admin auth views file
|
||||
"""
|
||||
# django imports
|
||||
from rest_framework.viewsets import GenericViewSet, mixins
|
||||
from rest_framework.viewsets import GenericViewSet
|
||||
from rest_framework.decorators import action
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
# local imports
|
||||
from account.utils import custom_response
|
||||
from base.constants import USER_TYPE
|
||||
from base.messages import SUCCESS_CODE, ERROR_CODE
|
||||
from web_admin.permission import AdminPermission
|
||||
from web_admin.serializers.auth_serializer import AdminForgotPasswordSerializer, AdminVerifyOTPSerializer
|
||||
from base.messages import SUCCESS_CODE
|
||||
from web_admin.serializers.auth_serializer import (AdminOTPSerializer, AdminVerifyOTPSerializer,
|
||||
AdminCreatePasswordSerializer)
|
||||
|
||||
USER = get_user_model()
|
||||
|
||||
|
||||
class ForgetAndResetPasswordViewSet(GenericViewSet):
|
||||
class ForgotAndResetPasswordViewSet(GenericViewSet):
|
||||
"""
|
||||
to reset admin password
|
||||
"""
|
||||
queryset = None
|
||||
|
||||
@action(methods=['post'], url_name='forgot-password', url_path='forgot-password',
|
||||
detail=False, serializer_class=AdminForgotPasswordSerializer)
|
||||
def admin_forgot_password(self, request):
|
||||
@action(methods=['post'], url_name='otp', url_path='otp',
|
||||
detail=False, serializer_class=AdminOTPSerializer)
|
||||
def admin_otp(self, request):
|
||||
"""
|
||||
api method to send otp
|
||||
:return: success message
|
||||
"""
|
||||
serializer = self.serializer_class(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
serializer.save()
|
||||
@ -30,14 +36,24 @@ class ForgetAndResetPasswordViewSet(GenericViewSet):
|
||||
@action(methods=['post'], url_name='verify-otp', url_path='verify-otp',
|
||||
detail=False, serializer_class=AdminVerifyOTPSerializer)
|
||||
def admin_verify_otp(self, request):
|
||||
"""
|
||||
api method to verify otp
|
||||
:return: success message
|
||||
"""
|
||||
serializer = self.serializer_class(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
return custom_response(SUCCESS_CODE['3011'])
|
||||
|
||||
@action(methods=['post'], url_name='resend-otp', url_path='resend-otp',
|
||||
detail=False, serializer_class=AdminForgotPasswordSerializer)
|
||||
def admin_resend_otp(self, request):
|
||||
@action(methods=['post'], url_name='create-password', url_path='create-password',
|
||||
detail=False, serializer_class=AdminCreatePasswordSerializer)
|
||||
def admin_create_password(self, request):
|
||||
"""
|
||||
api method to create new password
|
||||
:return: success message
|
||||
"""
|
||||
serializer = self.serializer_class(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
serializer.save()
|
||||
return custom_response(SUCCESS_CODE['3015'])
|
||||
user = USER.objects.filter(email=serializer.validated_data.get('email')).first()
|
||||
user.set_password(serializer.validated_data.get('new_password'))
|
||||
user.save()
|
||||
return custom_response(SUCCESS_CODE['3007'])
|
||||
|
||||
Reference in New Issue
Block a user