from collections import defaultdict 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, Role, 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, get_ip from rest_framework.exceptions import ValidationError from api.translation import __ class AdminCreate(AdminApiBase): def validate_request(self,request): rules = { "email": "required|not_exists:api.User,email", "name": "required", "password": "required", "password_confirmation": "required", "role": "nullable|array|exists:api.Role,id", "permissions": "nullable|array", "is_only_self": "nullable|boolean", } # if need rename field, can put here attribute = { # "email": __("attr.email"), } validator = BaseValidator(request.data, rules, attribute) error = validator.validate() if error: raise ValidationError(error) 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)} }) permissions = request.data.get("permissions", []) if permissions is None: permissions = [] if not isinstance(permissions, list): raise ValidationError({ "error": __('validation.invalid_format'), "errors": {"permissions": __("validation.invalid_format")} }) 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 True def post(self, request): # Validate input try: self.validate_request(request) old_new_data = { "new_data": None, } action="admin_create" # this is for audit log only encrypted_password = request.data.get('password') password = decrypt_aes_128_cbc(encrypted_password) role_ids = request.data.get("role", []) permissions = request.data.get("permissions", []) if permissions is None: permissions = [] with transaction.atomic(): admin = User.objects.create_user( name=request.data['name'], email=request.data['email'], password=password, user_type=User.USER_TYPE.get("admin"), status=User.STATUS_TO_CODE.get("active"), is_only_self=request.data.get("is_only_self", False), created_by=request.user.id, ) if role_ids: admin.role.set(role_ids) 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) # old_new_data['new_data'] = get_add_new_data(admin) # new_ids = request.data.get("role", []) # old_new_data = self.update_m2m_field(admin,'role',None,new_ids,old_new_data) # permissions = request.data.get("permissions", []) # if permissions: # UserPermission.bulk_insert(admin.id, permissions) # old_new_data.setdefault("new_data", [])["permission"]= [__("permission.admin." + p) for p in permissions] # 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) # ) # remark_param = { # "name":admin.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'], # ) 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)