From cf32e33064ff257b9ca14c9dbd32ad06f2bb2838 Mon Sep 17 00:00:00 2001 From: wxg0103 <727495428@qq.com> Date: Fri, 7 Aug 2026 10:17:08 +0800 Subject: [PATCH] style: format code for consistency and readability in user.py --- apps/users/views/user.py | 551 +++++++++++++++++++++++---------------- 1 file changed, 327 insertions(+), 224 deletions(-) diff --git a/apps/users/views/user.py b/apps/users/views/user.py index 5151bb80c66..bbe971f2c03 100644 --- a/apps/users/views/user.py +++ b/apps/users/views/user.py @@ -1,11 +1,12 @@ # coding=utf-8 """ - @project: MaxKB - @Author：虎虎 - @file： user.py - @date：2025/4/14 19:25 - @desc: +@project: MaxKB +@Author：虎虎 +@file： user.py +@date：2025/4/14 19:25 +@desc: """ + import json from django.core.cache import cache @@ -27,22 +28,39 @@ from common.utils.rsa_util import decrypt from maxkb.const import CONFIG from models_provider.api.model import DefaultModelResponse -from users.api.user import UserProfileAPI, TestWorkspacePermissionUserApi, DeleteUserApi, EditUserApi, \ - ChangeUserPasswordApi, UserPageApi, UserListApi, UserPasswordResponse, WorkspaceUserAPI, ResetPasswordAPI, \ - SendEmailAPI, CheckCodeAPI, SwitchUserLanguageAPI +from users.api.user import ( + UserProfileAPI, + TestWorkspacePermissionUserApi, + DeleteUserApi, + EditUserApi, + ChangeUserPasswordApi, + UserPageApi, + UserListApi, + UserPasswordResponse, + WorkspaceUserAPI, + ResetPasswordAPI, + SendEmailAPI, + CheckCodeAPI, + SwitchUserLanguageAPI, +) from users.models import User -from users.serializers.user import UserProfileSerializer, UserManageSerializer, CheckCodeSerializer, \ - SendEmailSerializer, RePasswordSerializer, SwitchLanguageSerializer, ResetCurrentUserPassword +from users.serializers.user import ( + UserProfileSerializer, + UserManageSerializer, + CheckCodeSerializer, + SendEmailSerializer, + RePasswordSerializer, + SwitchLanguageSerializer, + ResetCurrentUserPassword, +) -default_password = CONFIG.get('DEFAULT_PASSWORD', 'MaxKB@123..') +default_password = CONFIG.get("DEFAULT_PASSWORD", "MaxKB@123..") def get_user_operation_object(user_id): user_model = QuerySet(model=User).filter(id=user_id).first() if user_model is not None: - return { - "name": user_model.username - } + return {"name": user_model.username} return {} @@ -50,27 +68,24 @@ def get_re_password_details(request): path = request.path body = request.data query = request.query_params - body_copy = dict(body) if hasattr(body, 'items') else body + body_copy = dict(body) if hasattr(body, "items") else body if isinstance(body_copy, dict): - body_copy.pop('password', None) - body_copy.pop('re_password', None) - return { - "path": path, - "body": body_copy, - "query": query - } + body_copy.pop("password", None) + body_copy.pop("re_password", None) + return {"path": path, "body": body_copy, "query": query} class UserProfileView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get current user information"), - description=_("Get current user information"), - operation_id=_("Get current user information"), # type: ignore - tags=[_("User Management")], # type: ignore - - responses=UserProfileAPI.get_response()) + @extend_schema( + methods=["GET"], + summary=_("Get current user information"), + description=_("Get current user information"), + operation_id=_("Get current user information"), # type: ignore + tags=[_("User Management")], # type: ignore + responses=UserProfileAPI.get_response(), + ) def get(self, request: Request): return result.success(UserProfileSerializer().profile(request.user.profile, request.auth)) @@ -78,12 +93,14 @@ def get(self, request: Request): class TestPermissionsUserView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get current user information"), - description=_("Get current user information"), - operation_id="测试", - tags=[_("User Management")], # type: ignore - responses=UserProfileAPI.get_response()) + @extend_schema( + methods=["GET"], + summary=_("Get current user information"), + description=_("Get current user information"), + operation_id="测试", + tags=[_("User Management")], # type: ignore + responses=UserProfileAPI.get_response(), + ) @has_permissions(PermissionConstants.USER_EDIT, RoleConstants.ADMIN) def get(self, request: Request): return result.success(UserProfileSerializer().profile(request.user.profile, request.auth)) @@ -92,32 +109,35 @@ def get(self, request: Request): class SwitchUserLanguageView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['POST'], - summary=_("Switch Language"), - description=_("Switch Language"), - operation_id=_("Switch Language"), # type: ignore - tags=[_("User Management")], # type: ignore - request=SwitchUserLanguageAPI.get_request(), - ) - @log(menu='User management', operate='Switch Language', - get_operation_object=lambda r, k: {'name': r.user.username}) - @has_permissions(PermissionConstants.SWITCH_LANGUAGE, RoleConstants.ADMIN, RoleConstants.USER, - RoleConstants.WORKSPACE_MANAGE) + @extend_schema( + methods=["POST"], + summary=_("Switch Language"), + description=_("Switch Language"), + operation_id=_("Switch Language"), # type: ignore + tags=[_("User Management")], # type: ignore + request=SwitchUserLanguageAPI.get_request(), + ) + @log(menu="User management", operate="Switch Language", get_operation_object=lambda r, k: {"name": r.user.username}) + @has_permissions( + PermissionConstants.SWITCH_LANGUAGE, RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.WORKSPACE_MANAGE + ) def post(self, request: Request): - data = {**request.data, 'user_id': request.user.id} + data = {**request.data, "user_id": request.user.id} return result.success(SwitchLanguageSerializer(data=data).switch()) class TestWorkspacePermissionUserView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary="针对工作空间下权限校验", - description="针对工作空间下权限校验", - operation_id="针对工作空间下权限校验", - tags=[_("User Management")], # type: ignore - responses=UserProfileAPI.get_response(), - parameters=TestWorkspacePermissionUserApi.get_parameters()) + @extend_schema( + methods=["GET"], + summary="针对工作空间下权限校验", + description="针对工作空间下权限校验", + operation_id="针对工作空间下权限校验", + tags=[_("User Management")], # type: ignore + responses=UserProfileAPI.get_response(), + parameters=TestWorkspacePermissionUserApi.get_parameters(), + ) @has_permissions(PermissionConstants.USER_EDIT.get_workspace_permission(), RoleConstants.ADMIN) def get(self, request: Request, workspace_id): return result.success(UserProfileSerializer().profile(request.user.profile, request.auth)) @@ -126,49 +146,73 @@ def get(self, request: Request, workspace_id): class UserList(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get all user"), - description=_("Get all user"), - operation_id=_("Get all user"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=UserListApi.get_parameters(), - responses=UserListApi.get_response()) - @has_permissions(RoleConstants.WORKSPACE_MANAGE, RoleConstants.ADMIN, RoleConstants.EXTENDS_ADMIN, - RoleConstants.EXTENDS_WORKSPACE_MANAGE, RoleConstants.USER, RoleConstants.EXTENDS_USER) + @extend_schema( + methods=["GET"], + summary=_("Get all user"), + description=_("Get all user"), + operation_id=_("Get all user"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=UserListApi.get_parameters(), + responses=UserListApi.get_response(), + ) + @has_permissions( + RoleConstants.WORKSPACE_MANAGE, + RoleConstants.ADMIN, + RoleConstants.EXTENDS_ADMIN, + RoleConstants.EXTENDS_WORKSPACE_MANAGE, + RoleConstants.USER, + RoleConstants.EXTENDS_USER, + ) def get(self, request: Request): - nick_name = request.query_params.get('nick_name', None) + nick_name = request.query_params.get("nick_name", None) return result.success(UserManageSerializer().get_all_user_list(nick_name)) class WorkspaceUserListView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get user list under workspace"), - description=_("Get user list under workspace"), - operation_id=_("Get user list under workspace"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=WorkspaceUserAPI.get_parameters(), - responses=WorkspaceUserAPI.get_response()) - @has_permissions(RoleConstants.WORKSPACE_MANAGE, RoleConstants.ADMIN, RoleConstants.EXTENDS_ADMIN, - RoleConstants.EXTENDS_WORKSPACE_MANAGE, RoleConstants.USER, RoleConstants.EXTENDS_USER) + @extend_schema( + methods=["GET"], + summary=_("Get user list under workspace"), + description=_("Get user list under workspace"), + operation_id=_("Get user list under workspace"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=WorkspaceUserAPI.get_parameters(), + responses=WorkspaceUserAPI.get_response(), + ) + @has_permissions( + RoleConstants.WORKSPACE_MANAGE, + RoleConstants.ADMIN, + RoleConstants.EXTENDS_ADMIN, + RoleConstants.EXTENDS_WORKSPACE_MANAGE, + RoleConstants.USER, + RoleConstants.EXTENDS_USER, + ) def get(self, request: Request, workspace_id): - nick_name = request.query_params.get('nick_name', None) - return result.success(UserManageSerializer().get_user_list(str(request.user.id),workspace_id, nick_name)) + nick_name = request.query_params.get("nick_name", None) + return result.success(UserManageSerializer().get_user_list(str(request.user.id), workspace_id, nick_name)) class WorkspaceUserMemberView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get user member under workspace"), - description=_("Get user member under workspace"), - operation_id=_("Get user member under workspace"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=WorkspaceUserAPI.get_parameters(), - responses=WorkspaceUserAPI.get_response()) - @has_permissions(RoleConstants.WORKSPACE_MANAGE, RoleConstants.ADMIN, RoleConstants.EXTENDS_ADMIN, - RoleConstants.EXTENDS_WORKSPACE_MANAGE, RoleConstants.USER, RoleConstants.EXTENDS_USER) + @extend_schema( + methods=["GET"], + summary=_("Get user member under workspace"), + description=_("Get user member under workspace"), + operation_id=_("Get user member under workspace"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=WorkspaceUserAPI.get_parameters(), + responses=WorkspaceUserAPI.get_response(), + ) + @has_permissions( + RoleConstants.WORKSPACE_MANAGE, + RoleConstants.ADMIN, + RoleConstants.EXTENDS_ADMIN, + RoleConstants.EXTENDS_WORKSPACE_MANAGE, + RoleConstants.USER, + RoleConstants.EXTENDS_USER, + ) def get(self, request: Request, workspace_id): return result.success(UserManageSerializer().get_user_members(workspace_id)) @@ -176,162 +220,206 @@ def get(self, request: Request, workspace_id): class UserManage(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['POST'], - summary=_("Create user"), - description=_("Create user"), - operation_id=_("Create user"), # type: ignore - tags=[_("User Management")], # type: ignore - request=UserProfileAPI.get_request(), - responses=UserProfileAPI.get_response()) + @extend_schema( + methods=["POST"], + summary=_("Create user"), + description=_("Create user"), + operation_id=_("Create user"), # type: ignore + tags=[_("User Management")], # type: ignore + request=UserProfileAPI.get_request(), + responses=UserProfileAPI.get_response(), + ) @has_permissions(PermissionConstants.USER_CREATE, RoleConstants.ADMIN) - @log(menu='User management', operate='Add user', - get_operation_object=lambda r, k: {'name': r.data.get('username', None)}) + @log( + menu="User management", + operate="Add user", + get_operation_object=lambda r, k: {"name": r.data.get("username", None)}, + ) def post(self, request: Request): return result.success(UserManageSerializer().save(request.data, str(request.user.id))) class Password(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['Get'], - summary=_("Get default password"), - description=_("Get default password"), - operation_id=_("Get default password"), # type: ignore - tags=[_("User Management")], # type: ignore - responses=UserPasswordResponse.get_response()) - @has_permissions(PermissionConstants.USER_CREATE, PermissionConstants.CHAT_USER_CREATE, - PermissionConstants.WORKSPACE_CHAT_USER_CREATE, RoleConstants.ADMIN, - RoleConstants.WORKSPACE_MANAGE) + @extend_schema( + methods=["Get"], + summary=_("Get default password"), + description=_("Get default password"), + operation_id=_("Get default password"), # type: ignore + tags=[_("User Management")], # type: ignore + responses=UserPasswordResponse.get_response(), + ) + @has_permissions( + PermissionConstants.USER_CREATE, + PermissionConstants.CHAT_USER_CREATE, + RoleConstants.ADMIN, + RoleConstants.WORKSPACE_MANAGE, + ) def get(self, request: Request): - return result.success(data={'password': default_password}) + return result.success(data={"password": default_password}) class Operate(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['DELETE'], - description=_("Delete user"), - summary=_("Delete user"), - operation_id=_("Delete user"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=DeleteUserApi.get_parameters(), - responses=DefaultModelResponse.get_response()) + @extend_schema( + methods=["DELETE"], + description=_("Delete user"), + summary=_("Delete user"), + operation_id=_("Delete user"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=DeleteUserApi.get_parameters(), + responses=DefaultModelResponse.get_response(), + ) @has_permissions(PermissionConstants.USER_DELETE, RoleConstants.ADMIN) - @log(menu='User management', operate='Delete user', - get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id'))) + @log( + menu="User management", + operate="Delete user", + get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), + ) def delete(self, request: Request, user_id): - return result.success(UserManageSerializer.Operate(data={'id': user_id}).delete(with_valid=True)) - - @extend_schema(methods=['GET'], - summary=_("Get user information"), - description=_("Get user information"), - operation_id=_("Get user information"), # type: ignore - tags=[_("User Management")], # type: ignore - request=DeleteUserApi.get_parameters(), - responses=UserProfileAPI.get_response()) + return result.success(UserManageSerializer.Operate(data={"id": user_id}).delete(with_valid=True)) + + @extend_schema( + methods=["GET"], + summary=_("Get user information"), + description=_("Get user information"), + operation_id=_("Get user information"), # type: ignore + tags=[_("User Management")], # type: ignore + request=DeleteUserApi.get_parameters(), + responses=UserProfileAPI.get_response(), + ) @has_permissions(PermissionConstants.USER_READ, RoleConstants.ADMIN) def get(self, request: Request, user_id): - return result.success(UserManageSerializer.Operate(data={'id': user_id}).one(with_valid=True)) - - @extend_schema(methods=['PUT'], - summary=_("Update user information"), - description=_("Update user information"), - operation_id=_("Update user information"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=DeleteUserApi.get_parameters(), - request=EditUserApi.get_request(), - responses=UserProfileAPI.get_response()) + return result.success(UserManageSerializer.Operate(data={"id": user_id}).one(with_valid=True)) + + @extend_schema( + methods=["PUT"], + summary=_("Update user information"), + description=_("Update user information"), + operation_id=_("Update user information"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=DeleteUserApi.get_parameters(), + request=EditUserApi.get_request(), + responses=UserProfileAPI.get_response(), + ) @has_permissions(PermissionConstants.USER_EDIT, RoleConstants.ADMIN) - @log(menu='User management', operate='Update user information', - get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id'))) + @log( + menu="User management", + operate="Update user information", + get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), + ) def put(self, request: Request, user_id): return result.success( - UserManageSerializer.Operate(data={'id': user_id}).edit(request.data, str(request.user.id), - with_valid=True)) + UserManageSerializer.Operate(data={"id": user_id}).edit( + request.data, str(request.user.id), with_valid=True + ) + ) class BatchDelete(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['POST'], - description=_("Batch delete user"), - summary=_("Batch delete user"), - operation_id=_("Batch delete user"), # type: ignore - tags=[_("User Management")], # type: ignore - request=DeleteUserApi.get_request(), - responses=DefaultModelResponse.get_response()) + @extend_schema( + methods=["POST"], + description=_("Batch delete user"), + summary=_("Batch delete user"), + operation_id=_("Batch delete user"), # type: ignore + tags=[_("User Management")], # type: ignore + request=DeleteUserApi.get_request(), + responses=DefaultModelResponse.get_response(), + ) @has_permissions(PermissionConstants.USER_DELETE, RoleConstants.ADMIN) - @log(menu='User management', operate='Batch delete user', - get_operation_object=lambda r, k: get_user_operation_object(r.data.get('ids', []))) + @log( + menu="User management", + operate="Batch delete user", + get_operation_object=lambda r, k: get_user_operation_object(r.data.get("ids", [])), + ) def post(self, request: Request): - return result.success(UserManageSerializer.BatchDelete({'ids': request.data}).batch_delete(with_valid=True)) + return result.success(UserManageSerializer.BatchDelete({"ids": request.data}).batch_delete(with_valid=True)) class RePassword(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['PUT'], - summary=_("Change password"), - description=_("Change password"), - operation_id=_("Change password"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=DeleteUserApi.get_parameters(), - request=ChangeUserPasswordApi.get_request(), - responses=DefaultModelResponse.get_response()) - @log(menu='User management', operate='Change password', - get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id')), - get_details=get_re_password_details) + @extend_schema( + methods=["PUT"], + summary=_("Change password"), + description=_("Change password"), + operation_id=_("Change password"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=DeleteUserApi.get_parameters(), + request=ChangeUserPasswordApi.get_request(), + responses=DefaultModelResponse.get_response(), + ) + @log( + menu="User management", + operate="Change password", + get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), + get_details=get_re_password_details, + ) @has_permissions(PermissionConstants.USER_EDIT, RoleConstants.ADMIN) def put(self, request: Request, user_id): return result.success( - UserManageSerializer.Operate(data={'id': user_id}).re_password(request.data, with_valid=True)) + UserManageSerializer.Operate(data={"id": user_id}).re_password(request.data, with_valid=True) + ) class Page(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['GET'], - summary=_("Get user paginated list"), - description=_("Get user paginated list"), - operation_id=_("Get user paginated list"), # type: ignore - tags=[_("User Management")], # type: ignore - parameters=UserPageApi.get_parameters(), - responses=UserPageApi.get_response()) + @extend_schema( + methods=["GET"], + summary=_("Get user paginated list"), + description=_("Get user paginated list"), + operation_id=_("Get user paginated list"), # type: ignore + tags=[_("User Management")], # type: ignore + parameters=UserPageApi.get_parameters(), + responses=UserPageApi.get_response(), + ) @has_permissions(PermissionConstants.USER_READ, RoleConstants.ADMIN) def get(self, request: Request, current_page, page_size): - d = UserManageSerializer.Query( - data={**query_params_to_single_dict(request.query_params)}) + d = UserManageSerializer.Query(data={**query_params_to_single_dict(request.query_params)}) return result.success(d.page(current_page, page_size, str(request.user.id))) class RePasswordView(APIView): - - @extend_schema(methods=['POST'], - summary=_("Change password"), - description=_("Change password"), - operation_id=_("Change password"), # type: ignore - tags=[_("User Management")], # type: ignore - request=ResetPasswordAPI.get_request(), - responses=DefaultModelResponse.get_response()) - @log(menu='User management', operate='Change password', - get_operation_object=lambda r, k: {'name': r.user.username}, - get_details=get_re_password_details) + @extend_schema( + methods=["POST"], + summary=_("Change password"), + description=_("Change password"), + operation_id=_("Change password"), # type: ignore + tags=[_("User Management")], # type: ignore + request=ResetPasswordAPI.get_request(), + responses=DefaultModelResponse.get_response(), + ) + @log( + menu="User management", + operate="Change password", + get_operation_object=lambda r, k: {"name": r.user.username}, + get_details=get_re_password_details, + ) def post(self, request: Request): request_data = request.data if request_data.get("encrypted", False): - request_data['password'] = decrypt(request_data.get('password')) - request_data['re_password'] = decrypt(request_data.get('re_password')) + request_data["password"] = decrypt(request_data.get("password")) + request_data["re_password"] = decrypt(request_data.get("re_password")) serializer_obj = RePasswordSerializer(data=request_data) return result.success(serializer_obj.reset_password()) class SendEmail(APIView): - - @extend_schema(methods=['POST'], - summary=_("Send email"), - description=_("Send email"), - operation_id=_("Send email"), # type: ignore - tags=[_("User Management")], # type: ignore - request=SendEmailAPI.get_request(), - responses=SendEmailAPI.get_response()) - @log(menu='User management', operate='Send email', - get_operation_object=lambda r, k: {'name': r.data.get('email', None)}, - get_user=lambda r: {'user_name': None, 'email': r.data.get('email', None)}) + @extend_schema( + methods=["POST"], + summary=_("Send email"), + description=_("Send email"), + operation_id=_("Send email"), # type: ignore + tags=[_("User Management")], # type: ignore + request=SendEmailAPI.get_request(), + responses=SendEmailAPI.get_response(), + ) + @log( + menu="User management", + operate="Send email", + get_operation_object=lambda r, k: {"name": r.data.get("email", None)}, + get_user=lambda r: {"user_name": None, "email": r.data.get("email", None)}, + ) def post(self, request: Request): serializer_obj = SendEmailSerializer(data=request.data) if serializer_obj.is_valid(raise_exception=True): @@ -339,17 +427,21 @@ def post(self, request: Request): class CheckCode(APIView): - - @extend_schema(methods=['POST'], - summary=_("Check whether the verification code is correct"), - description=_("Check whether the verification code is correct"), - operation_id=_("Check whether the verification code is correct"), # type: ignore - tags=[_("User Management")], # type: ignore - request=CheckCodeAPI.get_request(), - responses=CheckCodeAPI.get_response()) - @log(menu='User management', operate='Check whether the verification code is correct', - get_operation_object=lambda r, k: {'name': r.data.get('email', None)}, - get_user=lambda r: {'user_name': None, 'email': r.data.get('email', None)}) + @extend_schema( + methods=["POST"], + summary=_("Check whether the verification code is correct"), + description=_("Check whether the verification code is correct"), + operation_id=_("Check whether the verification code is correct"), # type: ignore + tags=[_("User Management")], # type: ignore + request=CheckCodeAPI.get_request(), + responses=CheckCodeAPI.get_response(), + ) + @log( + menu="User management", + operate="Check whether the verification code is correct", + get_operation_object=lambda r, k: {"name": r.data.get("email", None)}, + get_user=lambda r: {"user_name": None, "email": r.data.get("email", None)}, + ) def post(self, request: Request): return result.success(CheckCodeSerializer(data=request.data).is_valid(raise_exception=True)) @@ -357,17 +449,22 @@ def post(self, request: Request): class SendEmailToCurrentUserView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['POST'], - summary=_("Send email to current user"), - description=_("Send email to current user"), - operation_id=_("Send email to current user"), # type: ignore - tags=[_("User Management")], # type: ignore - request=SendEmailAPI.get_request(), - responses=SendEmailAPI.get_response()) - @log(menu='User management', operate='Send email to current user', - get_operation_object=lambda r, k: {'name': r.user.username}) + @extend_schema( + methods=["POST"], + summary=_("Send email to current user"), + description=_("Send email to current user"), + operation_id=_("Send email to current user"), # type: ignore + tags=[_("User Management")], # type: ignore + request=SendEmailAPI.get_request(), + responses=SendEmailAPI.get_response(), + ) + @log( + menu="User management", + operate="Send email to current user", + get_operation_object=lambda r, k: {"name": r.user.username}, + ) def post(self, request: Request): - serializer_obj = SendEmailSerializer(data={'email': request.user.email, 'type': "reset_password"}) + serializer_obj = SendEmailSerializer(data={"email": request.user.email, "type": "reset_password"}) if serializer_obj.is_valid(raise_exception=True): return result.success(serializer_obj.send()) @@ -375,18 +472,24 @@ def post(self, request: Request): class ResetCurrentUserPasswordView(APIView): authentication_classes = [TokenAuth] - @extend_schema(methods=['POST'], - summary=_("Modify current user password"), - description=_("Modify current user password"), - operation_id=_("Modify current user password"), # type: ignore - tags=[_("User Management")], # type: ignore - request=ResetPasswordAPI.get_request(), - responses=DefaultModelResponse.get_response()) - @log(menu='User management', operate='Modify current user password', - get_operation_object=lambda r, k: {'name': r.user.username}, - get_details=get_re_password_details) - @has_permissions(PermissionConstants.CHANGE_PASSWORD, RoleConstants.ADMIN, RoleConstants.USER, - RoleConstants.WORKSPACE_MANAGE) + @extend_schema( + methods=["POST"], + summary=_("Modify current user password"), + description=_("Modify current user password"), + operation_id=_("Modify current user password"), # type: ignore + tags=[_("User Management")], # type: ignore + request=ResetPasswordAPI.get_request(), + responses=DefaultModelResponse.get_response(), + ) + @log( + menu="User management", + operate="Modify current user password", + get_operation_object=lambda r, k: {"name": r.user.username}, + get_details=get_re_password_details, + ) + @has_permissions( + PermissionConstants.CHANGE_PASSWORD, RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.WORKSPACE_MANAGE + ) def post(self, request: Request): request_data = request.data encrypted_data = request_data.get("encryptedData", "") @@ -397,7 +500,7 @@ def post(self, request: Request): decrypted_data = json.loads(decrypted_raw) if decrypted_raw else {} if isinstance(decrypted_data, dict): request_data = decrypted_data - except Exception as e: + except Exception: raise AppApiException(500, _("Invalid encrypted data")) serializer_obj = ResetCurrentUserPassword(data=request_data) if serializer_obj.reset_password(request.user.id):