mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2025-07-15 10:05:21 +00:00
279 lines
13 KiB
Python
279 lines
13 KiB
Python
"""Views of Guardian"""
|
|
"""Third party Django app"""
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework import viewsets, status
|
|
from rest_framework.pagination import PageNumberPagination
|
|
from django.contrib.auth.models import User
|
|
from django.utils import timezone
|
|
"""Import Django app"""
|
|
|
|
# Import guardian's model,
|
|
# Import junior's model,
|
|
# Import account's model,
|
|
# Import constant from
|
|
# base package,
|
|
# Import messages from
|
|
# base package,
|
|
# Import some functions
|
|
# from utils file
|
|
# Import account's serializer
|
|
# Import account's task
|
|
|
|
from .serializers import (UserSerializer, CreateGuardianSerializer, TaskSerializer, TaskDetailsSerializer,
|
|
TopJuniorSerializer, ApproveJuniorSerializer, ApproveTaskSerializer)
|
|
from .models import Guardian, JuniorTask
|
|
from junior.models import Junior, JuniorPoints
|
|
from account.models import UserEmailOtp, UserNotification
|
|
from .tasks import generate_otp
|
|
from account.utils import send_otp_email
|
|
from account.utils import custom_response, custom_error_response
|
|
from base.messages import ERROR_CODE, SUCCESS_CODE
|
|
from base.constants import NUMBER
|
|
from .utils import upload_image_to_alibaba
|
|
|
|
""" Define APIs """
|
|
# Define Signup API,
|
|
# update guardian profile,
|
|
# list of all task
|
|
# list of task according to the status of the task
|
|
# create task API
|
|
# search task by name of the task API
|
|
# top junior API,
|
|
# approve junior API
|
|
# approve task API
|
|
# Create your views here.
|
|
class SignupViewset(viewsets.ModelViewSet):
|
|
"""Signup view set"""
|
|
queryset = User.objects.all()
|
|
serializer_class = UserSerializer
|
|
def create(self, request, *args, **kwargs):
|
|
"""Create user profile"""
|
|
if request.data['user_type'] in ['1', '2']:
|
|
serializer = UserSerializer(context=request.data['user_type'], data=request.data)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
"""Generate otp"""
|
|
otp = generate_otp()
|
|
expiry = timezone.now() + timezone.timedelta(days=1)
|
|
# create user email otp object
|
|
UserEmailOtp.objects.create(email=request.data['email'], otp=otp,
|
|
user_type=str(request.data['user_type']), expired_at=expiry)
|
|
"""Send email to the register user"""
|
|
send_otp_email(request.data['email'], otp)
|
|
return custom_response(SUCCESS_CODE['3001'], {"email_otp": otp},
|
|
response_status=status.HTTP_200_OK)
|
|
return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST)
|
|
else:
|
|
return custom_error_response(ERROR_CODE['2028'], response_status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
class UpdateGuardianProfile(viewsets.ViewSet):
|
|
"""Update guardian profile"""
|
|
queryset = Guardian.objects.all()
|
|
serializer_class = CreateGuardianSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
"""Create guardian profile"""
|
|
data = request.data
|
|
image = request.data.get('image')
|
|
image_url = ''
|
|
if 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}"
|
|
# upload image on ali baba
|
|
image_url = upload_image_to_alibaba(image, filename)
|
|
data = {"image":image_url}
|
|
serializer = CreateGuardianSerializer(context={"user":request.user,
|
|
"first_name":request.data.get('first_name'),
|
|
"last_name": request.data.get('last_name'),
|
|
"image":image_url},
|
|
data=data)
|
|
if serializer.is_valid():
|
|
"""save serializer"""
|
|
serializer.save()
|
|
return custom_response(None, serializer.data,response_status=status.HTTP_200_OK)
|
|
return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
class AllTaskListAPIView(viewsets.ModelViewSet):
|
|
"""Update guardian profile"""
|
|
serializer_class = TaskDetailsSerializer
|
|
queryset = JuniorTask.objects.all()
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
"""Create guardian profile"""
|
|
queryset = JuniorTask.objects.filter(guardian__user=request.user)
|
|
# use TaskDetailsSerializer serializer
|
|
serializer = TaskDetailsSerializer(queryset, many=True)
|
|
return custom_response(None, serializer.data, response_status=status.HTTP_200_OK)
|
|
|
|
class TaskListAPIView(viewsets.ModelViewSet):
|
|
"""Update guardian profile"""
|
|
serializer_class = TaskDetailsSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
pagination_class = PageNumberPagination
|
|
queryset = JuniorTask.objects.all()
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
"""Create guardian profile"""
|
|
status_value = self.request.GET.get('status')
|
|
search = self.request.GET.get('search')
|
|
if search and str(status_value) == '0':
|
|
queryset = JuniorTask.objects.filter(guardian__user=request.user,
|
|
task_name__icontains=search).order_by('due_date', 'created_at')
|
|
elif search and str(status_value) != '0':
|
|
queryset = JuniorTask.objects.filter(guardian__user=request.user,task_name__icontains=search,
|
|
task_status=status_value).order_by('due_date', 'created_at')
|
|
if search is None and str(status_value) == '0':
|
|
queryset = JuniorTask.objects.filter(guardian__user=request.user).order_by('due_date', 'created_at')
|
|
elif search is None and str(status_value) != '0':
|
|
queryset = JuniorTask.objects.filter(guardian__user=request.user,
|
|
task_status=status_value).order_by('due_date','created_at')
|
|
paginator = self.pagination_class()
|
|
# use Pagination
|
|
paginated_queryset = paginator.paginate_queryset(queryset, request)
|
|
# use TaskDetailsSerializer serializer
|
|
serializer = TaskDetailsSerializer(paginated_queryset, many=True)
|
|
return custom_response(None, serializer.data, response_status=status.HTTP_200_OK)
|
|
|
|
class CreateTaskAPIView(viewsets.ModelViewSet):
|
|
"""create task for junior"""
|
|
serializer_class = TaskSerializer
|
|
queryset = JuniorTask.objects.all()
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
image = request.data['default_image']
|
|
data = request.data
|
|
if 'https' in str(image):
|
|
image_data = image
|
|
else:
|
|
filename = f"images/{image}"
|
|
if image and image.size == NUMBER['zero']:
|
|
return custom_error_response(ERROR_CODE['2035'], response_status=status.HTTP_400_BAD_REQUEST)
|
|
image_url = upload_image_to_alibaba(image, filename)
|
|
image_data = image_url
|
|
data.pop('default_image')
|
|
# use TaskSerializer serializer
|
|
serializer = TaskSerializer(context={"user":request.user, "image":image_data}, data=data)
|
|
if serializer.is_valid():
|
|
# save serializer
|
|
serializer.save()
|
|
return custom_response(SUCCESS_CODE['3018'], serializer.data, response_status=status.HTTP_200_OK)
|
|
return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
class SearchTaskListAPIView(viewsets.ModelViewSet):
|
|
"""Update guardian profile"""
|
|
serializer_class = TaskDetailsSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
pagination_class = PageNumberPagination
|
|
queryset = JuniorTask.objects.all()
|
|
|
|
def get_queryset(self):
|
|
"""Get the queryset for the view"""
|
|
title = self.request.GET.get('title')
|
|
# fetch junior query
|
|
junior_queryset = JuniorTask.objects.filter(guardian__user=self.request.user, task_name__icontains=title)\
|
|
.order_by('due_date', 'created_at')
|
|
return junior_queryset
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
"""Create guardian profile"""
|
|
queryset = self.get_queryset()
|
|
|
|
paginator = self.pagination_class()
|
|
# use pagination
|
|
paginated_queryset = paginator.paginate_queryset(queryset, request)
|
|
# use TaskSerializer serializer
|
|
serializer = TaskDetailsSerializer(paginated_queryset, many=True)
|
|
return custom_response(None, serializer.data, response_status=status.HTTP_200_OK)
|
|
|
|
|
|
class TopJuniorListAPIView(viewsets.ModelViewSet):
|
|
"""Top juniors list"""
|
|
serializer_class = TopJuniorSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
queryset = JuniorPoints.objects.all()
|
|
|
|
def get_serializer_context(self):
|
|
context = super().get_serializer_context()
|
|
context.update({'view': self})
|
|
return context
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
"""Fetch junior list of those who complete their tasks"""
|
|
junior_total_points = self.get_queryset().order_by('-total_task_points')
|
|
|
|
# Update the position field for each JuniorPoints object
|
|
for index, junior in enumerate(junior_total_points):
|
|
junior.position = index + 1
|
|
junior.save()
|
|
|
|
serializer = self.get_serializer(junior_total_points, many=True)
|
|
return custom_response(None, serializer.data, response_status=status.HTTP_200_OK)
|
|
|
|
|
|
class ApproveJuniorAPIView(viewsets.ViewSet):
|
|
"""approve junior by guardian"""
|
|
serializer_class = ApproveJuniorSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_queryset(self):
|
|
"""Get the queryset for the view"""
|
|
guardian = Guardian.objects.filter(user__email=self.request.user).last()
|
|
# fetch junior query
|
|
junior_queryset = Junior.objects.filter(id=self.request.data.get('junior_id')).last()
|
|
return guardian, junior_queryset
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
""" junior list"""
|
|
queryset = self.get_queryset()
|
|
# action 1 is use for approve and 2 for reject
|
|
if request.data['action'] == '1':
|
|
# use ApproveJuniorSerializer serializer
|
|
serializer = ApproveJuniorSerializer(context={"guardian_code": queryset[0].guardian_code,
|
|
"junior": queryset[1], "action": request.data['action']},
|
|
data=request.data)
|
|
if serializer.is_valid():
|
|
# save serializer
|
|
serializer.save()
|
|
return custom_response(SUCCESS_CODE['3023'], serializer.data, response_status=status.HTTP_200_OK)
|
|
else:
|
|
return custom_response(SUCCESS_CODE['3024'], response_status=status.HTTP_200_OK)
|
|
|
|
class ApproveTaskAPIView(viewsets.ViewSet):
|
|
"""approve junior by guardian"""
|
|
serializer_class = ApproveTaskSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_queryset(self):
|
|
"""Get the queryset for the view"""
|
|
guardian = Guardian.objects.filter(user__email=self.request.user).last()
|
|
# task query
|
|
task_queryset = JuniorTask.objects.filter(id=self.request.data.get('task_id'),
|
|
guardian=guardian,
|
|
junior=self.request.data.get('junior_id')).last()
|
|
return guardian, task_queryset
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
""" junior list"""
|
|
# action 1 is use for approve and 2 for reject
|
|
queryset = self.get_queryset()
|
|
# use ApproveJuniorSerializer serializer
|
|
serializer = ApproveTaskSerializer(context={"guardian_code": queryset[0].guardian_code,
|
|
"task_instance": queryset[1],
|
|
"action": str(request.data['action']),
|
|
"junior": self.request.data['junior_id']},
|
|
data=request.data)
|
|
if str(request.data['action']) == str(NUMBER['one']) and serializer.is_valid():
|
|
# save serializer
|
|
serializer.save()
|
|
return custom_response(SUCCESS_CODE['3025'], response_status=status.HTTP_200_OK)
|
|
elif str(request.data['action']) == str(NUMBER['two']) and serializer.is_valid():
|
|
# save serializer
|
|
serializer.save()
|
|
return custom_response(SUCCESS_CODE['3026'], response_status=status.HTTP_200_OK)
|
|
else:
|
|
return custom_response(ERROR_CODE['2038'], response_status=status.HTTP_400_BAD_REQUEST)
|