from datetime import date, datetime, time from django.utils import timezone from rest_framework import status from rest_framework.exceptions import ValidationError from api.models.audit_trail import AuditTrail from api.models.audit_trail_details import AuditTrailDetails from api.models.es_exam_requirement import EsExamRequirement from api.translation import __ from api.utils import log_critical_error, get_exception_detail, get_ip from api.validator import BaseValidator from api.views.admin.base import AdminApiBase class ExamRequirementUpdate(AdminApiBase): @staticmethod def serialize_audit_value(value): if isinstance(value, datetime): return value.isoformat(sep=" ") if isinstance(value, (date, time)): return value.isoformat() return value def parse_datetime_value(self, value, field): if value in (None, ""): return None if isinstance(value, datetime): return value.replace(tzinfo=None) datetime_formats = ( "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%dT%H:%M:%S", ) for dt_format in datetime_formats: try: return datetime.strptime(str(value), dt_format) except (ValueError, TypeError): pass message = __("validation.invalid", field=__(f"attr.{field}")) raise ValidationError({ "error": message, "errors": { field: [message], }, }) def validate_request(self, request): rules = { "id": "required|array|exists:api.EsExamRequirement,id", } # if need rename field, can put here attribute = { "description": __("attr.desc"), } # validate id first validator = BaseValidator(request.data, rules, attribute) error = validator.validate() if error: raise ValidationError(error) edit_obj = ( EsExamRequirement.objects .filter(id__in=request.data.get("id")) ) obj = edit_obj.first() rules = { "name": "nullable", "exam_period_id": "nullable|exists:api.EsExamPeriod,id", "description": "nullable", "planned_size": "nullable|integer|gte:0", "writing_time": "nullable|time", "fixed_start_date": "nullable|date", "reading_time": "nullable|time", "fixed_start_time": "nullable|time", "location_required": "nullable", "exclusive_use": "nullable", "minimum_split_size": "nullable|integer|gte:0", "earliest_start": "nullable", "latest_end": "nullable", "invigilator_suitability": "nullable|array|exists:api.EsSuitability,id", "location_suitability": "nullable|array|exists:api.EsSuitability,id", } if request.data.get("code") and edit_obj.count() == 1 and request.data.get("code") != obj.code: rules["code"] = "nullable|not_exists:api.EsExamRequirement,code" validator = BaseValidator(request.data, rules, attribute) error = validator.validate() if error: raise ValidationError(error) # Validate datetime format and range using final values after update. if "earliest_start" in request.data or "latest_end" in request.data: for current_obj in edit_obj: current_earliest_start = request.data.get("earliest_start") if "earliest_start" in request.data else current_obj.earliest_start current_latest_end = request.data.get("latest_end") if "latest_end" in request.data else current_obj.latest_end earliest_start = self.parse_datetime_value(current_earliest_start, "earliest_start") latest_end = self.parse_datetime_value(current_latest_end, "latest_end") if earliest_start and latest_end and earliest_start > latest_end: message = __("validation.gt.date", field=__("attr.latest_end"), value=str(current_earliest_start)) raise ValidationError({ "error": message, "errors": { "latest_end": message, }, }) return edit_obj def post(self, request): # Validate input try: edit_obj = self.validate_request(request) action = "exam_requirement_update" # this is for audit log only old_new_data = { "new_data": {}, "old_data": {}, } updated = False update_fields = {} # if more than 1 record edit, ignore code if edit_obj.count() == 1: current_edit_obj = edit_obj.first() if "code" in request.data: old_new_data.setdefault("old_data", {}).setdefault(current_edit_obj.id, {})["code"] = current_edit_obj.code old_new_data.setdefault("new_data", {}).setdefault(current_edit_obj.id, {})["code"] = request.data.get("code") update_fields["code"] = request.data["code"] updated = True ignore_fields = ["id", "timestamp", "signature"] valid_fields = {f.column for f in EsExamRequirement._meta.fields} for key, val in request.data.items(): if key not in ignore_fields and key in valid_fields: if key == "code" and edit_obj.count() != 1: continue updated = True update_fields[key] = val if updated and update_fields: for obj in edit_obj: for key, val in update_fields.items(): old_val = getattr(obj, key) serialized_old_val = self.serialize_audit_value(old_val) serialized_new_val = self.serialize_audit_value(val) if str(serialized_old_val) != str(serialized_new_val): old_new_data.setdefault("old_data", {}).setdefault(obj.id, {})[key] = serialized_old_val old_new_data.setdefault("new_data", {}).setdefault(obj.id, {})[key] = serialized_new_val update_fields["updated_by"] = request.user.id update_fields["updated_at"] = timezone.now() edit_obj.update(**update_fields) if "invigilator_suitability" in request.data: new_ids = request.data.get("invigilator_suitability", []) old_new_data = self.update_m2m_field_bulk(edit_obj, "invigilator_suitability", new_ids, old_new_data) if "location_suitability" in request.data: new_ids = request.data.get("location_suitability", []) old_new_data = self.update_m2m_field_bulk(edit_obj, "location_suitability", new_ids, old_new_data) if old_new_data["old_data"] or old_new_data["new_data"]: # insert audit trail and details # can insert the parent first, then only insert details audit_trail = AuditTrail.objects.create( user_id=request.user.id, type=self.audit_type, ip_address=get_ip(request), ) code = ",".join(edit_obj.values_list("code", flat=True)) remark_param = { "code": code, } # call to a function for insert details, so in future if want change to used worker, can modify in function only AuditTrailDetails.custom_insert( audit_trail=audit_trail, action=action, remark_param=remark_param, new_data=old_new_data["new_data"], old_data=old_new_data["old_data"], ) response = { } 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)