from django.contrib.auth.hashers import make_password from rest_framework import status from api.admin_permission_conf import PERMISSIONS from api.models import User, UserPermission from api.models.audit_trail import AuditTrail from api.models.audit_trail_details import AuditTrailDetails from api.views.admin.base import AdminApiBase from api.validator import BaseValidator from api.utils import decrypt_aes_128_cbc, log_critical_error, get_exception_detail, get_ip, get_multiple_old_new_data 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", } # 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" # if need rename field, can put here attribute = { # "email": __("attr.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: 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", []) invalid_permission = [p for p in permissions if p not in PERMISSIONS] if invalid_permission: raise ValidationError({ "error": __('validation.exists'), "errors": { "permissions": __("validation.exists", field=__("attr.permissions"), values=invalid_permission)} }) 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'] # 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": 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 updated and update_fields: update_fields['updated_by'] = request.user.id edit_admin.update(**update_fields) # 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)