from rest_framework import status from rest_framework.exceptions import ValidationError from django.db.models import Count from api.models import EsInvigilator, EsSuitability from api.translation import __ from api.utils import log_critical_error, get_exception_detail from api.validator import BaseValidator from api.views.admin.base import AdminApiBase class InvigilatorList(AdminApiBase): list_default_order_column = "id" def validate_request(self, request): rules = { "filter": "nullable", "page": "nullable", "per_page": "nullable", "sort_by": "nullable", "order_by": "nullable", } validator = BaseValidator(request.data, rules, {}) error = validator.validate() if error: raise ValidationError(error) def custom_listing_filter(self, request, model): filters = request.data.get("filter") if filters: for key, val in filters.items(): if val == "" or val is None: continue match key: case "desc" | "email" | "phone": filter_kwargs = {f"{key}__icontains": val} model = model.filter(**filter_kwargs) case "status": if isinstance(val, (list, tuple)): model = model.filter(status__in=val) else: model = model.filter(status=val) case "tag_id": if isinstance(val, (list, tuple)): model = model.filter(tag__tag_id__in=val) else: model = model.filter(tag__tag_id=val) case "role_id": if isinstance(val, (list, tuple)): model = model.filter(role__in=val) else: model = model.filter(role__id=val) case "suitability": if isinstance(val, (list, tuple)): model = model.filter(suitability__in=val) else: model = model.filter(suitability__id=val) case "engagement_date_from" | "termination_date_from": db_field = "engagement_date" if key == "engagement_date_from" else "termination_date" to_key = f"{db_field}_to" if not filters.get(to_key): model = model.filter(**{f"{db_field}": val}) else: model = model.filter(**{f"{db_field}__gte": val}) case "engagement_date_to" | "termination_date_to": db_field = "engagement_date" if key == "engagement_date_to" else "termination_date" from_key = f"{db_field}_from" if not filters.get(from_key): model = model.filter(**{f"{db_field}": val}) else: model = model.filter(**{f"{db_field}__lte": val}) return model def post(self, request): try: self.api_log_skip_outgoing_data = True self.validate_request(request) data_details = self.get_data(request) response = { "page": data_details["page"], "per_page": data_details["per_page"], "total": data_details["total"], "data": data_details["data"] } return self.api_response(data=response) except ValidationError as e: first_message = e.detail['error'] errors = e.detail['errors'] return self.api_response(error=first_message,errors=errors,code=status.HTTP_400_BAD_REQUEST) except Exception as e: e_details = get_exception_detail(e) log_critical_error(user_id=None,descr=e_details['descr'],url=e_details['url'],trace=e_details['trace']) return self.api_response(error=__("message.internal_server_error"),code=status.HTTP_500_INTERNAL_SERVER_ERROR) def invigilator_listing_pagination(self, request, model): page, per_page = self.parse_pagination(request) sort_by = request.data.get("sort_by") order_by = request.data.get("order_by") model = model.distinct() total = model.count() if sort_by: if sort_by in ["role", "tag", "suitability"]: sort_col = f"_sort_{sort_by}" model = model.annotate(**{sort_col: Count(sort_by, distinct=True)}) else: sort_col = sort_by if str(order_by).lower() == "desc": ordering_string = f"-{sort_col}" else: ordering_string = sort_col model = model.order_by(ordering_string) else: model = model.order_by(self.list_default_order_column) if per_page != -1: start_index = (page - 1) * per_page end_index = start_index + per_page paginated_data = model[start_index:end_index] else: paginated_data = model return { "total": total, "paginated_data": paginated_data, "page": page, "per_page": per_page, } def get_data(self,request): Invigilators = ( EsInvigilator.objects .select_related("department") .prefetch_related( "suitability", "role", "tag__tag" #tags means the relation add in models, __tag means the relation table tag_id get the tag object ) ) Invigilators = self.listing_filter(request, Invigilators) Invigilators = self.custom_listing_filter(request, Invigilators) paginated_details = self.invigilator_listing_pagination(request, Invigilators) total = paginated_details["total"] paginated_data = paginated_details["paginated_data"] page = paginated_details["page"] per_page = paginated_details["per_page"] return_data=[] for val in paginated_data: data={} data["id"] = val.id data["code"] = val.code data["name"] = val.name data['desc'] = val.desc data['department'] = val.department.name if val.department else None data['department_id'] = val.department_id data['email'] = val.email data['phone'] = val.phone data['engagement_date'] = val.engagement_date data['termination_date'] = val.termination_date data["status"] = val.status data["status_text"] = __("attr.invigilator_status." + str(val.status)) data['role'] = list( val.role.values("id","name","color") ) data['suitability'] = [ {"id": suitability.id, "name": suitability.name} for suitability in val.suitability.all() ] data["tag"] = [ {"id": tag.tag.id, "name": tag.tag.name} for tag in val.tag.all() ] for i in range(1,11): extra_data_field = f"extra_data_{i}" data[extra_data_field] = getattr(val,extra_data_field,None) return_data.append(data) return { "page": page, "per_page": per_page, "total": total, "data": return_data }