from collections import defaultdict from django.contrib.auth.hashers import make_password from django.db import transaction from rest_framework import serializers, status from api.admin_permission_conf import PERMISSIONS from api.models import User, UserPermission, UserPermissionDepartment, TtDepartment 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 AdminUpdate(AdminApiBase): def validate_request(self,request): rules = { "id": "required|array|exists:api.User,id", } # if need rename field, can put here attribute = { # "email": __("attr.email"), } # validate id first, cause need get admin by id, if wrong id direct return error validator = BaseValidator(request.data,rules,attribute) error = validator.validate() if error: raise ValidationError(error) edit_admin = ( User.objects .prefetch_related( "role", 'userpermission_set' # tags means the relation add in models, __tag means the relation table tag_id get the tag object #2 ) .filter(id__in=request.data['id'], user_type=User.USER_TYPE['admin']) ) rules = { "name": "nullable", "password": "nullable", "password_confirmation": "nullable", "role": "nullable|array|exists:api.Role,id", "permissions": "nullable|array", "is_only_self": "nullable|boolean", } # not allow update if request.data.get("email") and edit_admin.count() == 1: admin = edit_admin.first() if request.data.get("email") != admin.email: rules['email'] = "nullable|not_exists:api.User,email" validator = BaseValidator(request.data, rules, attribute) error = validator.validate() # custom error after basic validation if error: raise ValidationError(error) if "password" in request.data and request.data.get("password") not in [None, ""]: encrypted_password_confirmation = request.data.get('password_confirmation') encrypted_password = request.data.get('password') try: password = decrypt_aes_128_cbc(encrypted_password) password_confirmation = decrypt_aes_128_cbc(encrypted_password_confirmation) except Exception: raise ValidationError({ "error": __('validation.invalid_format'), "errors": {"password": __("validation.invalid_format")} }) if password != password_confirmation: raise ValidationError({ "error": __('validation.not_match', field=__("attr.password"), confirm_field=__("attr.password_confirmation")), "errors": {"password": __('validation.not_match', field=__("attr.password"), confirm_field=__("attr.password_confirmation"))} }) if len(password) < 8: raise ValidationError({ "error": __('validation.min_string', min=8), "errors": {"password": __('validation.min_string', min=8)} }) if "permissions" in request.data: permissions = request.data.get("permissions") if permissions is not None and not isinstance(permissions, list): raise ValidationError({ "error": __('validation.invalid_format'), "errors": {"permissions": __("validation.invalid_format")} }) if permissions is not None: valid_permission_codes = PERMISSIONS seen_permissions = set() for idx, item in enumerate(permissions): if not isinstance(item, dict): raise ValidationError({ "error": __('validation.invalid_format'), "errors": {f"permissions.{idx}": __("validation.invalid_format")} }) permission = item.get("permission") grant_to_all_department = item.get("grant_to_all_department") department_ids = item.get("department_ids", []) if permission not in valid_permission_codes: raise ValidationError({ "error": __('validation.exists'), "errors": { f"permissions.{idx}.permission": __("validation.exists", field="permission", values=[permission]) } }) if permission in seen_permissions: raise ValidationError({ "error": __('validation.duplicate'), "errors": { f"permissions.{idx}.permission": __("validation.duplicate") } }) seen_permissions.add(permission) if grant_to_all_department is None or not isinstance(grant_to_all_department, bool): raise ValidationError({ "error": __('validation.invalid_format'), "errors": { f"permissions.{idx}.grant_to_all_department": __("validation.invalid_format") } }) if department_ids is None: department_ids = [] if not isinstance(department_ids, list): raise ValidationError({ "error": __('validation.invalid_format'), "errors": { f"permissions.{idx}.department_ids": __("validation.invalid_format") } }) if not grant_to_all_department: exists_count = TtDepartment.objects.filter(id__in=department_ids).count() if exists_count != len(set(department_ids)): raise ValidationError({ "error": __('validation.exists'), "errors": { f"permissions.{idx}.department_ids": __("validation.exists", field="department_ids") } }) return edit_admin def post(self, request): # Validate input try: edit_admin = self.validate_request(request) action="admin_update" # this is for audit log only updated = False update_fields = {} # ignore_fields = ['id','timestamp','signature','permissions','role','password_confirmation'] ignore_fields = ['id', 'timestamp', 'signature', 'password_confirmation', 'permissions', 'role', 'is_only_self'] # if more than 1 admin edit, ignore email if edit_admin.count() > 1: ignore_fields.append('email') ignore_fields.append('password') valid_fields = {f.column for f in User._meta.fields} for key, val in request.data.items(): if key not in ignore_fields and key in valid_fields: updated = True match key: case "password": if val not in [None, ""]: password = decrypt_aes_128_cbc(request.data['password']) update_fields['password'] = make_password(password) case _: # default the key will same as db column name update_fields[key] = val # be4 save need get the old new data already, because bulk update is no matter got change or no changes also will update # old_new_data = get_multiple_old_new_data(edit_admin,request.data.items(),ignore_fields,valid_fields) if "is_only_self" in request.data: updated = True update_fields["is_only_self"] = request.data.get("is_only_self", False) with transaction.atomic(): if updated and update_fields: update_fields['updated_by'] = request.user.id edit_admin.update(**update_fields) if edit_admin.count() == 1: admin = edit_admin.first() if "role" in request.data: role_ids = request.data.get("role", []) if role_ids is None: role_ids = [] admin.role.set(role_ids) if "permissions" in request.data: permissions = request.data.get("permissions") if permissions is None: permissions = [] UserPermissionDepartment.objects.filter(user_permission__user=admin).delete() UserPermission.objects.filter(user=admin).delete() for item in permissions: permission_row = UserPermission.objects.create( user=admin, permission=item["permission"], grant_to_all_department=item["grant_to_all_department"], ) if not item["grant_to_all_department"]: department_ids = item.get("department_ids", []) department_objs = [ UserPermissionDepartment( user_permission=permission_row, department_id=department_id ) for department_id in department_ids ] if department_objs: UserPermissionDepartment.objects.bulk_create(department_objs) # new_ids change to set so it cam used to compare # if "role" in request.data: # new_ids = request.data.get("role", []) # old_new_data = self.update_m2m_field_bulk(edit_admin,'role',new_ids,old_new_data) # # if "permissions" in request.data: # permissions = request.data['permissions'] # for val in edit_admin: # current_permission = list(val.userpermission_set.values_list("permission",flat=True)) # if set(permissions) != set(current_permission): # val.userpermission_set.all().delete() # UserPermission.bulk_insert(val.id, permissions) # old_new_data.setdefault("old_data", []).setdefault(val.id,{})["permission"] = [__("permission.admin." + p) for p in current_permission] # old_new_data.setdefault("new_data", []).setdefault(val.id,{})["permission"] = [__("permission.admin." + p) for p in permissions] # if old_new_data['old_data'] or old_new_data['new_data'] or updated: # # 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) # ) # # name = ",".join(edit_admin.values_list("name", flat=True)) # remark_param = { # "name":name # } # # # 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)