from collections import defaultdict from rest_framework import serializers, status from api.models import TtStaffSuitability, TtDepartment, TtZone, TtAvailability, TtStartPreference, TtUsagePreference, \ TtSuitability, TtTag, TtSetting, TtPos, TtPosModuleGroup, TtPosModuleGroupModule, TtAcademicTerm, \ TtConstraintProfile from api.models.tt_student_set import TtStudentSet from api.views.admin.base import AdminApiBase from api.validator import BaseValidator from api.utils import decrypt_aes_128_cbc, encrypt_aes_128_cbc, log_critical_error, get_exception_detail from rest_framework.exceptions import ValidationError from api.translation import __ class StudentSetRules(AdminApiBase): def validate_request(self,request): rules = { # "id": "nullable|array|exists:api.TtStudentSet,id", "academic_term_id": "required|exists:api.TtAcademicTerm,id", } # if need rename field, can put here attribute = { # "email": __("attr.email"), } # validate id first, cause need get student_set by id, if wrong id direct return error validator = BaseValidator(request.data,rules,attribute) error = validator.validate() if error: raise ValidationError(error) # custom error after basic validation if error: raise ValidationError(error) def post(self, request): # Validate input try: self.api_log_skip_outgoing_data=True self.validate_request(request) ids = request.data.get("id") academic_term_id = request.data.get("academic_term_id") if not ids: ids = [] student_set = None if isinstance(ids, (list, tuple)) and len(ids) == 1: student_set = ( TtStudentSet.objects .select_related("department", "academic_term","pos", "availability", "start_preference", "usage_preference") .prefetch_related( "suitability", "constraint_profile", "pos_module_group_module", "tag__tag" # tags means the relation add in models, __tag means the relation table tag_id get the tag object #2 ) .filter(id__in=ids) .first() ) info = None if student_set: info = { "code": student_set.code, "name": student_set.name, "desc": student_set.desc, "department_id": student_set.department_id, "academic_term_id": student_set.academic_term_id, "planned_size": student_set.planned_size, "real_size": student_set.real_size, "module": student_set.module.all().values_list("id",flat=True), "availability_id": student_set.availability_id, "availability_pattern": student_set.availability_pattern, "usage_preference_id": student_set.usage_preference_id, "usage_preference_pattern": student_set.usage_preference_pattern, "start_preference_id": student_set.start_preference_id, "start_preference_pattern": student_set.start_preference_pattern, "tag": student_set.tag.all().values_list("tag_id",flat=True), } for i in range(1, 11): extra_data_field = f"extra_data_{i}" info[extra_data_field] = getattr(student_set, extra_data_field, None) models = { "department": (TtDepartment, ["id", "code", "name"]), "academic_term": (TtAcademicTerm,["id", "code", "name", "start_time", "end_time", "default_availability", "default_usage", "default_start"]), "availability": (TtAvailability, ["id", "code", "name", "pattern"]), "start_preference": (TtStartPreference, ["id", "code", "name", "pattern"]), "usage_preference": (TtUsagePreference, ["id", "code", "name", "pattern"]), "tag": (TtTag, ["id", "code", "name"]), } data = {} data['delete_confirmation_api'] = False data["allow_remove"] = [ "desc", "pos_id", "module", "department_id", "availability_id", "availability_pattern", "start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern", "tag", *[f"extra_data_{i}" for i in range(1, 11)] ] for key, (model, fields) in models.items(): data[key] = list( model.objects.filter(status=model.STATUS_TO_CODE['active']).values(*fields) ) """ Explanation of filter part "pos_module_group__pos__academic_term_id = academic_term_id" "__" double under scroll means connect to another table pos_module_group means first FK from pos_module_group_module pos means FK from pos_module_group academic_term_id means the column in pos table want to filter query will like SELECT * FROM tt_pos_module_group_module m JOIN tt_pos_module_group g ON m.pos_module_group_id = g.id JOIN tt_pos p ON g.pos_id = p.id WHERE p.academic_term_id = academic_term_id; """ pos_module_group_modules = ( TtPosModuleGroupModule.objects .filter(pos_module_group__pos__academic_term_id=academic_term_id) .select_related("pos_module_group", "pos_module_group__module_group","module") # pos_module_group__module_group means get the FK in pos_module_group ) # make the pos_module_group_modules data key by pos_id, so it can filter inside pos rules modules_by_pos = defaultdict(list) pos_used_module = defaultdict(set) used_module = set() data['module'] = [] for val in pos_module_group_modules: pos_id = val.pos_module_group.pos_id # Skip duplicate module_code for same POS if val.module.id in pos_used_module[pos_id]: continue pos_used_module[pos_id].add(val.module.id) modules_by_pos[pos_id].append({ # "pos_module_group_module_id": val.id, "module_id": val.module.id, "module_name": val.module.name, "module_code": val.module.code, # "module_group_name": val.pos_module_group.module_group.name, }) if val.module.id not in used_module: used_module.add(val.module.id) data['module'].append({ "id": val.module.id, "name": val.module.name, "code": val.module.code, }) related_pos = TtPos.objects.filter(academic_term_id=academic_term_id) data['pos'] = [] for pos in related_pos: data['pos'].append({ "id": pos.id, "code": pos.code, "name": pos.name, "modules": modules_by_pos.get(pos.id, []) }) setting_params = { "minute_per_slot", "slot_per_day", "slot_per_week", } data['based_setting'] = TtSetting.get_multiple_setting(setting_params) data["constraint_profile"] = TtConstraintProfile.get_constraint_profile_rules() response = { "info": info, "rules": 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)