from django.http import HttpResponse from rest_framework.views import APIView from rest_framework.response import Response from base.filters import (RotatingShiftAssignFilters, RotatingWorkTypeAssignFilter, ShiftRequestFilter, WorkTypeRequestFilter) from base.views import (is_reportingmanger, rotating_work_type_assign_export, shift_request_export, work_type_request_export) from ...api_decorators.base.decorators import (check_approval_status, manager_or_owner_permission_required, manager_permission_required) from employee.models import Actiontype, Employee from ...api_methods.base.methods import groupby_queryset, permission_based_queryset from horilla.decorators import permission_required from ...api_serializers.base.serializers import (ActiontypeSerializer, CompanySerializer, DepartmentSerializer, EmployeeShiftScheduleSerializer, EmployeeShiftSerializer, JobPositionSerializer, JobRoleSerializer, RotatingShiftAssignSerializer, RotatingShiftSerializer, RotatingWorkTypeAssignSerializer, RotatingWorkTypeSerializer, ShiftRequestSerializer, WorkTypeRequestSerializer, WorkTypeSerializer) from base.models import (Department, EmployeeShift, EmployeeShiftSchedule, JobPosition, JobRole, Company, RotatingShift, RotatingShiftAssign, RotatingWorkType, RotatingWorkTypeAssign, ShiftRequest, WorkType, WorkTypeRequest) from rest_framework.pagination import PageNumberPagination from rest_framework.permissions import IsAuthenticated from django_filters.rest_framework import DjangoFilterBackend from django.utils.decorators import method_decorator from notifications.signals import notify def object_check(cls, pk): try: obj = cls.objects.get(id=pk) return obj except cls.DoesNotExist: return None def object_delete(cls, pk): try: cls.objects.get(id=pk).delete() return "", 200 except Exception as e: return {"error": str(e)}, 400 def individual_permssion_check(request): employee_id = request.GET.get('employee_id') employee = Employee.objects.filter(id= employee_id).first() if request.user.employee_get == employee: return True elif employee.employee_work_info.reporting_manager_id == request.user.employee_get: return True elif request.user.has_perm('base.view_rotatingworktypeassign'): return True return False def _is_reportingmanger(request, instance): """ If the instance have employee id field then you can use this method to know the request user employee is the reporting manager of the instance """ manager = request.user.employee_get try: employee_work_info_manager = ( instance.employee_work_info.reporting_manager_id ) except Exception: return HttpResponse("This Employee Dont Have any work information") return manager == employee_work_info_manager class JobPositionView(APIView): serializer_class = JobPositionSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_jobposition")) def get(self, request, pk=None): if pk: job_position = object_check(JobPosition, pk) if job_position is None: return Response({"error": "Job position not found "}, status=404) serializer = self.serializer_class(job_position) return Response(serializer.data, status=200) job_positions = JobPosition.objects.all() paginater = PageNumberPagination() page = paginater.paginate_queryset(job_positions, request) serializer = self.serializer_class(page, many=True) return paginater.get_paginated_response(serializer.data) @method_decorator(permission_required("base.change_jobposition")) def put(self, request, pk): job_position = object_check(JobPosition, pk) if job_position is None: return Response({"error": "Job position not found "}, status=404) serializer = self.serializer_class(job_position, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.add_jobposition")) def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_jobposition")) def delete(self, request, pk): job_position = object_check(JobPosition, pk) if job_position is None: return Response({"error": "Job position not found "}, status=404) response, status_code = object_delete(JobPosition, pk) return Response(response, status=status_code) class DepartmentView(APIView): serializer_class = DepartmentSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_department"), name='dispatch') def get(self, request, pk=None): if pk: department = object_check(Department, pk) if department is None: return Response({"error": "Department not found "}, status=404) serializer = self.serializer_class(department) return Response(serializer.data, status=200) departments = Department.objects.all() paginator = PageNumberPagination() page = paginator.paginate_queryset(departments, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) @method_decorator(permission_required("base.change_department"), name='dispatch') def put(self, request, pk): department = object_check(Department, pk) if department is None: return Response({"error": "Department not found "}, status=404) serializer = self.serializer_class(department, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.add_department"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_department"), name='dispatch') def delete(self, request, pk): department = object_check(Department, pk) if department is None: return Response({"error": "Department not found "}, status=404) response, status_code = object_delete(Department, pk) return Response(response, status=status_code) class JobRoleView(APIView): serializer_class = JobRoleSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_jobrole"), name='dispatch') def get(self, request, pk=None): if pk: job_role = object_check(JobRole, pk) if job_role is None: return Response({"error": "Job role not found "}, status=404) serializer = self.serializer_class(job_role) return Response(serializer.data, status=200) job_roles = JobRole.objects.all() paginator = PageNumberPagination() page = paginator.paginate_queryset(job_roles, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) @method_decorator(permission_required("base.change_jobrole"), name='dispatch') def put(self, request, pk): job_role = object_check(JobRole, pk) if job_role is None: return Response({"error": "Job role not found "}, status=404) serializer = self.serializer_class(job_role, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.add_jobrole"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_jobrole"), name='dispatch') def delete(self, request, pk): job_role = object_check(JobRole, pk) if job_role is None: return Response({"error": "Job role not found "}, status=404) response, status_code = object_delete(JobRole, pk) return Response(response, status=status_code) class CompanyView(APIView): serializer_class = CompanySerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_company"), name='dispatch') def get(self, request, pk=None): if pk: company = object_check(Company, pk) if company is None: return Response({"error": "Company not found "}, status=404) serializer = self.serializer_class(company) return Response(serializer.data, status=200) companies = Company.objects.all() paginator = PageNumberPagination() page = paginator.paginate_queryset(companies, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) @method_decorator(permission_required("base.change_company"), name='dispatch') def put(self, request, pk): company = object_check(Company, pk) if company is None: return Response({"error": "Company not found "}, status=404) serializer = self.serializer_class(company, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.add_company"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_company"), name='dispatch') def delete(self, request, pk): company = object_check(Company, pk) if company is None: return Response({"error": "Company not found "}, status=400) response, status_code = object_delete(Company, pk) return Response(response, status=status_code) class WorkTypeView(APIView): serializer_class = WorkTypeSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if pk: work_type = object_check(WorkType, pk) if work_type is None: return Response({"error": "WorkType not found"}, status=404) serializer = self.serializer_class(work_type) return Response(serializer.data, status=200) work_types = WorkType.objects.all() serializer = self.serializer_class(work_types, many=True) return Response(serializer.data) @method_decorator(permission_required("base.add_worktype"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_worktype"), name='dispatch') def put(self, request, pk): work_type = object_check(WorkType, pk) if work_type is None: return Response({"error": "WorkType not found"}, status=404) serializer = self.serializer_class(work_type, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_worktype"), name='dispatch') def delete(self, request, pk): work_type = object_check(WorkType, pk) if work_type is None: return Response({"error": "WorkType not found"}, status=404) response, status_code = object_delete(WorkType, pk) return Response(response, status=status_code) class WorkTypeRequestView(APIView): serializer_class = WorkTypeRequestSerializer filterset_class = WorkTypeRequestFilter permission_classes = [IsAuthenticated] def get_queryset(self, request): queryset = WorkTypeRequest.objects.all() user = request.user # checking user level permissions perm = "base.view_worktyperequest" queryset = permission_based_queryset(user, perm, queryset,user_obj=True) return queryset def get(self, request, pk=None): # individual object workflow if pk: work_type_request = object_check(WorkTypeRequest, pk) if work_type_request is None: return Response({"error": "WorkTypeRequest not found"}, status=404) serializer = self.serializer_class(work_type_request) return Response(serializer.data, status=200) # permission based queryset work_type_requests = self.get_queryset(request) # filtering queryset work_type_request_filter_queryset = self.filterset_class( request.GET, queryset=work_type_requests).qs # groupby workflow field_name = request.GET.get("groupby_field", None) if field_name: url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, work_type_request_filter_queryset) # pagination workflow paginater = PageNumberPagination() page = paginater.paginate_queryset( work_type_request_filter_queryset, request) serializer = self.serializer_class(page, many=True) return paginater.get_paginated_response(serializer.data) def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): instance = serializer.save() try: notify.send( instance.employee_id, recipient=( instance.employee_id.employee_work_info.reporting_manager_id.employee_user_id ), verb=f"You have new work type request to \ validate for {instance.employee_id}", verb_ar=f"لديك طلب نوع وظيفة جديد للتحقق من \ {instance.employee_id}", verb_de=f"Sie haben eine neue Arbeitstypanfrage zur \ Validierung für {instance.employee_id}", verb_es=f"Tiene una nueva solicitud de tipo de trabajo para \ validar para {instance.employee_id}", verb_fr=f"Vous avez une nouvelle demande de type de travail\ à valider pour {instance.employee_id}", icon="information", redirect=f"/employee/work-type-request-view?id={instance.id}", api_redirect=f"/api/base/worktype-requests/{instance.id}", ) return Response(serializer.data, status=201) except Exception as E: return Response(serializer.errors, status=400) return Response(serializer.errors, status=400) @check_approval_status(WorkTypeRequest, "base.change_worktyperequest") @manager_or_owner_permission_required(WorkTypeRequest, "base.change_worktyperequest") def put(self, request, pk): work_type_request = object_check(WorkTypeRequest, pk) if work_type_request is None: return Response({"error": "WorkTypeRequest not found"}, status=404) serializer = self.serializer_class( work_type_request, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @check_approval_status(WorkTypeRequest, "base.change_worktyperequest") @manager_or_owner_permission_required(WorkTypeRequest, "base.delete_worktyperequest") def delete(self, request, pk): work_type_request = object_check(WorkTypeRequest, pk) if work_type_request is None: return Response({"error": "WorkTypeRequest not found"}, status=404) response, status_code = object_delete(WorkTypeRequest, pk) return Response(response, status=status_code) class WorkTypeRequestCancelView(APIView): def put(self,request,pk): work_type_request = WorkTypeRequest.find(pk) if ( is_reportingmanger(request, work_type_request) or request.user.has_perm("base.cancel_worktyperequest") or work_type_request.employee_id == request.user.employee_get and work_type_request.approved == False ): work_type_request.canceled = True work_type_request.approved = False work_type_request.employee_id.employee_work_info.work_type_id = ( work_type_request.previous_work_type_id ) work_type_request.employee_id.employee_work_info.save() work_type_request.save() try: notify.send( request.user.employee_get, recipient=work_type_request.employee_id.employee_user_id, verb="Your work type request has been rejected.", verb_ar="تم إلغاء طلب نوع وظيفتك", verb_de="Ihre Arbeitstypanfrage wurde storniert", verb_es="Su solicitud de tipo de trabajo ha sido cancelada", verb_fr="Votre demande de type de travail a été annulée", redirect=f"/employee/work-type-request-view?id={work_type_request.id}", icon="close", api_redirect = '/api/base/worktype-requests//' ) except : pass return Response(status=200) class WorkRequestApproveView(APIView): permission_classes =[IsAuthenticated] def put(self,request,pk): work_type_request = WorkTypeRequest.find(pk) if ( is_reportingmanger(request, work_type_request) or request.user.has_perm("approve_worktyperequest") or request.user.has_perm("change_worktyperequest") and not work_type_request.approved ): """ Here the request will be approved, can send mail right here """ if not work_type_request.is_any_work_type_request_exists(): work_type_request.approved = True work_type_request.canceled = False work_type_request.save() try: notify.send( request.user.employee_get, recipient=work_type_request.employee_id.employee_user_id, verb="Your work type request has been approved.", verb_ar="تمت الموافقة على طلب نوع وظيفتك.", verb_de="Ihre Arbeitstypanfrage wurde genehmigt.", verb_es="Su solicitud de tipo de trabajo ha sido aprobada.", verb_fr="Votre demande de type de travail a été approuvée.", redirect=f"/employee/work-type-request-view?id={work_type_request.id}", icon="checkmark", api_redirect ="/api/base/worktype-requests//" ) return Response({"status":"approved"}) except Exception as e: return Response({"error":str(e)},status=400) else: return Response({"error":"You don't have permission"},status=400) class WorkTypeRequestExport(APIView): permission_classes = [IsAuthenticated] @manager_permission_required("base.view_worktyperequest") def get(self, request): return work_type_request_export(request) class RotatingWorkTypeView(APIView): serializer_class = RotatingWorkTypeSerializer permission_classes = [IsAuthenticated] def get_queryset(self, request): queryset = RotatingWorkType.objects.all() user = request.user # checking user level permissions perm = "base.view_rotatingworktype" queryset = permission_based_queryset(user, perm, queryset) return queryset def get(self, request, pk=None): if pk: rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) serializer = self.serializer_class(rotating_work_type) return Response(serializer.data, status=200) rotating_work_types = self.get_queryset(request) serializer = self.serializer_class(rotating_work_types, many=True) return Response(serializer.data, status=200) @method_decorator(permission_required("base.add_rotatingworktype"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_rotatingworktype"), name='dispatch') def put(self, request, pk): rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) serializer = self.serializer_class( rotating_work_type, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_rotatingworktype"), name='dispatch') def delete(self, request, pk): rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) response, status_code = object_delete(RotatingWorkType, pk) return Response(response, status=status_code) class IndividualRotatingWorktypesView(APIView): serializer_class = RotatingWorkTypeAssignSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if individual_permssion_check(request) == False: return Response({'error':'you have no permssion to view'},status=400) if pk: rotating_work_type_assign = object_check( RotatingWorkTypeAssign, pk) if rotating_work_type_assign is None: return Response({"error": "RotatingWorkTypeAssign not found"}, status=404) serializer = self.serializer_class(rotating_work_type_assign) return Response(serializer.data, status=200) employee_id = request.GET.get("employee_id",None) rotating_work_type_assigns = RotatingWorkTypeAssign.objects.filter(employee_id=employee_id) pagenation = PageNumberPagination() page = pagenation.paginate_queryset( rotating_work_type_assigns, request) serializer = self.serializer_class( page, many=True) return pagenation.get_paginated_response(serializer.data) class RotatingWorkTypeAssignView(APIView): serializer_class = RotatingWorkTypeAssignSerializer filterset_class = RotatingWorkTypeAssignFilter permission_classes = [IsAuthenticated] @manager_permission_required("base.view_rotatingworktypeassign") def get(self, request, pk=None): if pk: rotating_work_type_assign = object_check( RotatingWorkTypeAssign, pk) if rotating_work_type_assign is None: return Response({"error": "RotatingWorkTypeAssign not found"}, status=404) serializer = self.serializer_class(rotating_work_type_assign) return Response(serializer.data, status=200) rotating_work_type_assigns = RotatingWorkTypeAssign.objects.all() rotating_work_type_assigns_filter_queryset = self.filterset_class( request.GET, queryset=rotating_work_type_assigns).qs field_name = request.GET.get("groupby_field", None) if field_name: # groupby workflow url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, rotating_work_type_assigns_filter_queryset) pagenation = PageNumberPagination() page = pagenation.paginate_queryset( rotating_work_type_assigns_filter_queryset, request) serializer = self.serializer_class( page, many=True) return pagenation.get_paginated_response(serializer.data) @manager_permission_required("base.add_rotatingworktypeassign") def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): obj = serializer.save() try: users = [employee.employee_user_id for employee in obj] notify.send( request.user.employee_get, recipient=users, verb="You are added to rotating work type", verb_ar="تمت إضافتك إلى نوع العمل المتناوب", verb_de="Sie werden zum rotierenden Arbeitstyp hinzugefügt", verb_es="Se le agrega al tipo de trabajo rotativo", verb_fr="Vous êtes ajouté au type de travail rotatif", icon="infinite", redirect="/employee/employee-profile/", api_redirect='' ) except: pass return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @manager_permission_required("base.change_rotatingworktypeassign") def put(self, request, pk): rotating_work_type_assign = object_check(RotatingWorkTypeAssign, pk) if rotating_work_type_assign is None: return Response({"error": "RotatingWorkTypeAssign not found"}, status=404) serializer = self.serializer_class( rotating_work_type_assign, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @manager_permission_required("base.delete_rotatingworktypeassign") def delete(self, request, pk): rotating_work_type_assign = object_check(RotatingWorkTypeAssign, pk) if rotating_work_type_assign is None: return Response({"error": "RotatingWorkTypeAssign not found"}, status=404) response, status_code = object_delete(RotatingWorkTypeAssign, pk) return Response(response, status=status_code) class IndividualWorkTypeRequestView(APIView): serializer_class = WorkTypeRequestSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if individual_permssion_check(request) == False: return Response({'error':'you have no permssion to view'},status=400) # individual object workflow if pk: work_type_request = object_check(WorkTypeRequest, pk) if work_type_request is None: return Response({"error": "WorkTypeRequest not found"}, status=404) serializer = self.serializer_class(work_type_request) return Response(serializer.data, status=200) employee_id = request.GET.get('employee_id',None) work_type_request = WorkTypeRequest.objects.filter(employee_id = employee_id) paginater = PageNumberPagination() page = paginater.paginate_queryset( work_type_request, request) serializer = self.serializer_class(page, many=True) return paginater.get_paginated_response(serializer.data) class EmployeeShiftView(APIView): serializer_class = EmployeeShiftSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if pk: employee_shift = object_check(EmployeeShift, pk) if employee_shift is None: return Response({"error": "EmployeeShift not found"}, status=404) serializer = self.serializer_class(employee_shift) return Response(serializer.data, status=200) employee_shifts = EmployeeShift.objects.all() serializer = self.serializer_class(employee_shifts, many=True) return Response(serializer.data, status=200) @method_decorator(permission_required("base.add_employeeshift"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_employeeshift"), name='dispatch') def put(self, request, pk): employee_shift = object_check(EmployeeShift, pk) if employee_shift is None: return Response({"error": "EmployeeShift not found"}, status=404) serializer = self.serializer_class(employee_shift, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_employeeshift"), name='dispatch') def delete(self, request, pk): employee_shift = object_check(EmployeeShift, pk) if employee_shift is None: return Response({"error": "EmployeeShift not found"}, status=404) response, status_code = object_delete(EmployeeShift, pk) return Response(response, status=status_code) class EmployeeShiftScheduleView(APIView): serializer_class = EmployeeShiftScheduleSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_employeeshiftschedule"), name='dispatch') def get(self, request, pk=None): if pk: employee_shift_schedule = object_check(EmployeeShiftSchedule, pk) if employee_shift_schedule is None: return Response({"error": "EmployeeShiftSchedule not found"}, status=404) serializer = self.serializer_class(employee_shift_schedule) return Response(serializer.data, status=200) employee_shift_schedules = EmployeeShiftSchedule.objects.all() serializer = self.serializer_class(employee_shift_schedules, many=True) return Response(serializer.data, status=200) @method_decorator(permission_required("base.add_employeeshiftschedule"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_employeeshiftschedule"), name='dispatch') def put(self, request, pk): employee_shift_schedule = object_check(EmployeeShiftSchedule, pk) if employee_shift_schedule is None: return Response({"error": "EmployeeShiftSchedule not found"}, status=404) serializer = self.serializer_class( employee_shift_schedule, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_employeeshiftschedule"), name='dispatch') def delete(self, request, pk): employee_shift_schedule = object_check(EmployeeShiftSchedule, pk) if employee_shift_schedule is None: return Response({"error": "EmployeeShiftSchedule not found"}, status=404) response, status_code = object_delete(EmployeeShiftSchedule, pk) return Response(response, status=status_code) class RotatingShiftView(APIView): serializer_class = RotatingShiftSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_rotatingshift"), name='dispatch') def get(self, request, pk=None): if pk: rotating_shift = object_check(RotatingShift, pk) if rotating_shift is None: return Response({"error": "RotatingShift not found"}, status=404) serializer = self.serializer_class(rotating_shift) return Response(serializer.data, status=200) employee_id = request.GET.get('employee_id') # Get the employee_id from query parameters if employee_id: # Check if employee_ids are present in the request rotating_shifts = RotatingShift.objects.filter(employee_id__in=[employee_id]) rotating_shifts = RotatingShift.objects.all() serializer = self.serializer_class(rotating_shifts, many=True) return Response(serializer.data, status=200) @method_decorator(permission_required("base.add_rotatingshift"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_rotatingshift"), name='dispatch') def put(self, request, pk): rotating_shift = object_check(RotatingShift, pk) if rotating_shift is None: return Response({"error": "RotatingShift not found"}, status=404) serializer = self.serializer_class(rotating_shift, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_rotatingshift"), name='dispatch') def delete(self, request, pk): rotating_shift = object_check(RotatingShift, pk) if rotating_shift is None: return Response({"error": "RotatingShift not found"}, status=404) response, status_code = object_delete(RotatingShift, pk) return Response(response, status=status_code) class IndividualRotatingShiftView(APIView): serializer_class = RotatingShiftAssignSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if individual_permssion_check(request) == False: return Response({'error':'you have no permssion to view'},status=400) if pk: rotating_shift_assign = object_check(RotatingShiftAssign, pk) if rotating_shift_assign is None: return Response({"error": "RotatingShiftAssign not found"}, status=404) serializer = self.serializer_class(rotating_shift_assign) return Response(serializer.data, status=200) employee_id = request.GET.get('employee_id',None) rotating_shift_assigns = RotatingShiftAssign.objects.filter(employee_id= employee_id) paginator = PageNumberPagination() page = paginator.paginate_queryset( rotating_shift_assigns, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) class RotatingShiftAssignView(APIView): serializer_class = RotatingShiftAssignSerializer filterset_class = RotatingShiftAssignFilters permission_classes = [IsAuthenticated] @manager_permission_required("base.view_rotatingshiftassign") def get(self, request, pk=None): if pk: rotating_shift_assign = object_check(RotatingShiftAssign, pk) if rotating_shift_assign is None: return Response({"error": "RotatingShiftAssign not found"}, status=404) serializer = self.serializer_class(rotating_shift_assign) return Response(serializer.data, status=200) rotating_shift_assigns = RotatingShiftAssign.objects.all() rotating_shift_assigns_filter_queryset = self.filterset_class( request.GET, queryset=rotating_shift_assigns).qs field_name = request.GET.get("groupby_field", None) if field_name: # groupby workflow url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, rotating_shift_assigns_filter_queryset) paginator = PageNumberPagination() page = paginator.paginate_queryset( rotating_shift_assigns_filter_queryset, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) @manager_permission_required("base.add_rotatingshiftassign") def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @manager_permission_required("base.change_rotatingshiftassign") def put(self, request, pk): rotating_shift_assign = object_check(RotatingShiftAssign, pk) if rotating_shift_assign is None: return Response({"error": "RotatingShiftAssign not found"}, status=404) serializer = self.serializer_class( rotating_shift_assign, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @manager_permission_required("base.delete_rotatingshiftassign") def delete(self, request, pk): rotating_shift_assign = object_check(RotatingShiftAssign, pk) if rotating_shift_assign is None: return Response({"error": "RotatingShiftAssign not found"}, status=404) response, status_code = object_delete(RotatingShiftAssign, pk) return Response(response, status=status_code) class IndividualShiftRequestView(APIView): serializer_class = ShiftRequestSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if individual_permssion_check(request) == False: return Response({'error':'you have no permssion to view'},status=400) if pk: shift_request = object_check(ShiftRequest, pk) if shift_request is None: return Response({"error": "EmployeeShift not found"}, status=404) serializer = self.serializer_class(shift_request) return Response(serializer.data, status=200) employee_id = request.GET.get('employee_id',None) shift_requests = ShiftRequest.objects.filter(employee_id = employee_id) paginater = PageNumberPagination() page = paginater.paginate_queryset( shift_requests, request) serializer = self.serializer_class(page, many=True) return paginater.get_paginated_response(serializer.data) class ShiftRequestView(APIView): serializer_class = ShiftRequestSerializer filter_backends = [DjangoFilterBackend] filterset_class = ShiftRequestFilter permission_classes = [IsAuthenticated] def get_queryset(self, request): queryset = ShiftRequest.objects.all() user = request.user # checking user level permissions perm = "base.view_shiftrequest" queryset = permission_based_queryset(user, perm, queryset,user_obj=True) return queryset def get(self, request, pk=None): # individual section if pk: shift_request = object_check(ShiftRequest, pk) if shift_request is None: return Response({"error": "ShiftRequest not found"}, status=404) serializer = self.serializer_class(shift_request) return Response(serializer.data, status=200) # filter section shift_requests = self.get_queryset(request) shift_requests_filter_queryset = self.filterset_class( request.GET, queryset=shift_requests).qs # groupby section field_name = request.GET.get("groupby_field", None) if field_name: url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, shift_requests_filter_queryset) # pagination section paginator = PageNumberPagination() page = paginator.paginate_queryset( shift_requests_filter_queryset, request) serializer = self.serializer_class(page, many=True) return paginator.get_paginated_response(serializer.data) def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @check_approval_status(ShiftRequest, "base.change_shiftrequest") @manager_or_owner_permission_required(ShiftRequest, "base.change_shiftrequest") def put(self, request, pk): shift_request = object_check(ShiftRequest, pk) if shift_request is None: return Response({"error": "ShiftRequest not found"}, status=404) serializer = self.serializer_class(shift_request, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @check_approval_status(ShiftRequest, "base.delete_shiftrequest") @manager_or_owner_permission_required(ShiftRequest, "base.delete_shiftrequest") def delete(self, request, pk): shift_request = object_check(ShiftRequest, pk) if shift_request is None: return Response({"error": "ShiftRequest not found"}, status=404) response, status_code = object_delete(ShiftRequest, pk) return Response(response, status=status_code) class RotatingWorkTypeView(APIView): serializer_class = RotatingWorkTypeSerializer permission_classes = [IsAuthenticated] @method_decorator(permission_required("base.view_rotatingworktype"), name='dispatch') def get(self, request, pk=None): if pk: rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) serializer = self.serializer_class(rotating_work_type) return Response(serializer.data, status=200) rotating_work_types = RotatingWorkType.objects.all() serializer = self.serializer_class(rotating_work_types, many=True) return Response(serializer.data, status=200) @method_decorator(permission_required("base.add_rotatingworktype"), name='dispatch') def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.change_rotatingworktype"), name='dispatch') def put(self, request, pk): rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) serializer = self.serializer_class( rotating_work_type, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("base.delete_rotatingworktype"), name='dispatch') def delete(self, request, pk): rotating_work_type = object_check(RotatingWorkType, pk) if rotating_work_type is None: return Response({"error": "RotatingWorkType not found"}, status=404) response, status_code = object_delete(RotatingWorkType, pk) return Response(response, status=status_code) class ShiftRequestApproveView(APIView): permission_classes = [IsAuthenticated] def put(self, request, pk): shift_request = ShiftRequest.objects.get(id=pk) if ( is_reportingmanger(request, shift_request) or request.user.has_perm("approve_shiftrequest") or request.user.has_perm("change_shiftrequest") and not shift_request.approved ): """ here the request will be approved, can send mail right here """ if not shift_request.is_any_request_exists(): shift_request.approved = True shift_request.canceled = False shift_request.save() return Response({"status": "success"}, status=200) else: return Response({"error": "Already request exits on same date"}, status=400) return Response({"error": "No permission "}, status=400) class ShiftRequestBulkApproveView(APIView): permission_classes = [IsAuthenticated] def post(self, request): ids = request.data["ids"] length = len(ids) count = 0 for id in ids: shift_request = ShiftRequest.objects.get(id=id) if ( is_reportingmanger(request, shift_request) or request.user.has_perm("approve_shiftrequest") or request.user.has_perm("change_shiftrequest") and not shift_request.approved ): """ here the request will be approved, can send mail right here """ shift_request.approved = True shift_request.canceled = False employee_work_info = shift_request.employee_id.employee_work_info employee_work_info.shift_id = shift_request.shift_id employee_work_info.save() shift_request.save() count += 1 if length == count: return Response({"status": "success"}, status=200) return Response({"status": "failed"}, status=400) class ShiftRequestCancelView(APIView): permission_classes = [IsAuthenticated] def post(self, request, pk): shift_request = ShiftRequest.objects.get(id=pk) if ( is_reportingmanger(request, shift_request) or request.user.has_perm("base.cancel_shiftrequest") or shift_request.employee_id == request.user.employee_get and shift_request.approved == False ): shift_request.canceled = True shift_request.approved = False shift_request.employee_id.employee_work_info.shift_id = ( shift_request.previous_shift_id ) shift_request.employee_id.employee_work_info.save() shift_request.save() return Response({"status": "success"}, status=200) return Response({"status": "failed"}, status=400) class ShiftRequestBulkCancelView(APIView): permission_classes = [IsAuthenticated] def post(self, request): ids = request.data.get("ids", None) length = len(ids) count = 0 for id in ids: shift_request = ShiftRequest.objects.get(id=id) if ( is_reportingmanger(request, shift_request) or request.user.has_perm("base.cancel_shiftrequest") or shift_request.employee_id == request.user.employee_get and shift_request.approved == False ): shift_request.canceled = True shift_request.approved = False shift_request.employee_id.employee_work_info.shift_id = ( shift_request.previous_shift_id ) shift_request.employee_id.employee_work_info.save() shift_request.save() count += 1 if length == count: return Response({"status": "success"}, status=200) return Response({"status": "failed"}, status=400) class ShiftRequestDeleteView(APIView): permission_classes = [IsAuthenticated] def delete(self, request, pk=None): if pk is None: try: ids = request.data["ids"] shift_requests = ShiftRequest.objects.filter(id__in=ids) shift_requests.delete() except Exception as e: return Response({"status": "failed", "error": str(e)}, status=400) return Response({"status": "success"}, status=200) try: shift_request = ShiftRequest.objects.get(id=pk) if not shift_request.approved: raise shift_request.delete() except ShiftRequest.DoesNotExist: return Response({"status": "failed", "error": "Shift request does not exists"}, status=400) return Response({"status": "deleted"}, status=200) class ShiftRequestExportView(APIView): permission_classes = [IsAuthenticated] def get(self, request): return shift_request_export(request) class ShiftRequestAllocationView(APIView): permission_classes = [IsAuthenticated] def post(self, request,id): shift_request = ShiftRequest.objects.get(id=id) if not shift_request.is_any_request_exists(): shift_request.reallocate_approved = True shift_request.reallocate_canceled = False shift_request.save() return Response({"status": "success"}, status=200) return Response({"status": "failed"}, status=400) class RotatingShiftAssignExport(APIView): permission_classes = [IsAuthenticated] def get(self, request): return rotating_work_type_assign_export(request) class RotatingShiftAssignBulkArchive(APIView): permission_classes = [IsAuthenticated] def put(self, request, status): ids = request.data.get('ids', None) try: rotating_shift_asssign = RotatingShiftAssign.objects.filter( id__in=ids) rotating_shift_asssign.update(is_active=status) return Response({"status": "success"}, status=200) except Exception as E: return Response({"error": str(E)}, status=400) class RotatingShiftAssignBulkDelete(APIView): permission_classes = [IsAuthenticated] def delete(self, request): ids = request.data.get('ids', None) try: rotating_shift_asssign = RotatingShiftAssign.objects.filter( id__in=ids) rotating_shift_asssign.delete() return Response({"status": "success"}, status=200) except Exception as E: return Response({"error": str(E)}, status=400) class ActiontypeView(APIView): serializer_class = ActiontypeSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if pk: action_type = object_check(Actiontype, pk) if action_type is None: return Response({"error": "Actiontype not found"}, status=404) serializer = self.serializer_class(action_type) return Response(serializer.data, status=200) action_types = Actiontype.objects.all() paginater = PageNumberPagination() page = paginater.paginate_queryset(action_types, request) serializer = self.serializer_class(page, many=True) return paginater.get_paginated_response(serializer.data) def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def put(self, request, pk): action_type = object_check(Actiontype, pk) if action_type is None: return Response({"error": "Actiontype not found"}, status=404) serializer = self.serializer_class( action_type, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) def delete(self, request, pk): action_type = object_check(Actiontype, pk) if action_type is None: return Response({"error": "Actiontype not found"}, status=404) response, status_code = object_delete(Actiontype, pk) return Response(response, status=status_code) class RotatingWorKTypePermissionCheck(APIView): permission_classes = [IsAuthenticated] def get(self,request,id): manager = Employee.objects.filter(id=id).first().get_reporting_manager() if request.user.has_perm('base.add_rotatingworktypeassign') or request.user.employee_get == manager : return Response(status=200) return Response(status=400) class RotatingShiftPermissionCheck(APIView): permission_classes = [IsAuthenticated] def get(self,request,id): manager = Employee.objects.filter(id=id).first().get_reporting_manager() if request.user.has_perm('base.add_rotatingshiftassign') or request.user.employee_get == manager : return Response(status=200) return Response(status=400) class WorktypeRequestApprovePermissionCheck(APIView): permission_classes = [IsAuthenticated] def get(self,request): instance = request.user.employee_get if ( _is_reportingmanger(request, instance) or request.user.has_perm("approve_shiftrequest") or request.user.has_perm("change_shiftrequest") ): return Response(status=200) return Response(status=400) class ShiftRequestApprovePermissionCheck(APIView): permission_classes = [IsAuthenticated] def get(self,request): instance = request.user.employee_get if ( _is_reportingmanger(request, instance) or request.user.has_perm("approve_shiftrequest") or request.user.has_perm("change_shiftrequest") ): return Response(status=200) return Response(status=400) class EmployeeTabPermissionCheck(APIView): permission_classes = [IsAuthenticated] def get(self,request): instance = request.user.employee_get if ( _is_reportingmanger(request, instance) or request.user.has_perms(["attendance.view_worktyperequest","perms.attendance.view_shiftrequest"]) ): return Response(status=200) return Response(status=400)