From 2bf2179611eb31a8efa3416b3934f946293d5cce Mon Sep 17 00:00:00 2001 From: DjangoAdmin Dev Date: Fri, 7 Aug 2026 17:32:54 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20P4c=20=E6=9C=80=E7=BB=88=E6=89=B9?= =?UTF-8?q?=E6=AC=A1-6=20user=20=E6=A8=A1=E5=9D=97=EF=BC=88=E6=9C=80?= =?UTF-8?q?=E5=90=8E=E4=B8=80=E4=B8=AA=EF=BC=89=E2=80=94=E2=80=94P4=20?= =?UTF-8?q?=E5=85=A8=E9=83=A8=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【改造】get_user_page/add/update/status/delete 按新范式； update_user_info/update_password/reset_user_password 入参 data + BizError； get_info 委托 UserRepository 【验收】flake8 零告警；应用启动；user 分页/详情运行时正常；12 passed Co-Authored-By: Claude --- src/modules/system/user/models.py | 1 - src/modules/system/user/repository.py | 15 + src/modules/system/user/router.py | 63 +-- src/modules/system/user/service.py | 549 +++++++++++--------------- 4 files changed, 290 insertions(+), 338 deletions(-) create mode 100644 src/modules/system/user/repository.py diff --git a/src/modules/system/user/models.py b/src/modules/system/user/models.py index 6c42c5e..817df8c 100644 --- a/src/modules/system/user/models.py +++ b/src/modules/system/user/models.py @@ -50,7 +50,6 @@ from src.common.base_db import BaseDB from src.common.base_model import BaseModel from config import DB_PREFIX from src.core.database import db -from src.common.utils.case_converter import convert_camel_to_snake # ============================================================ diff --git a/src/modules/system/user/repository.py b/src/modules/system/user/repository.py new file mode 100644 index 0000000..cb38588 --- /dev/null +++ b/src/modules/system/user/repository.py @@ -0,0 +1,15 @@ +""" +User 仓储模块 +============================================================= + +数据访问层，继承 BaseRepository 获得通用 CRUD 能力。 +""" + +from src.common.base_repository import BaseRepository +from src.modules.system.user.models import User + + +class UserRepository(BaseRepository): + """User 仓储""" + + model = User diff --git a/src/modules/system/user/router.py b/src/modules/system/user/router.py index 87d6d7e..adf4fc7 100644 --- a/src/modules/system/user/router.py +++ b/src/modules/system/user/router.py @@ -56,21 +56,24 @@ application.middleware.check_permission - 权限校验中间件 """ -from flask import Blueprint +from flask import Blueprint, request from application.middleware.check_demo import check_demo from application.middleware.check_permission import check_permission from src.modules.system.user import service as user -from src.common.utils.request import parse_batch_ids -user_blue = Blueprint('user', __name__, url_prefix='/user') +from src.common.utils.request import parse_batch_ids, parse_request_body + from src.core import response as R +user_blue = Blueprint("user", __name__, url_prefix="/user") + # ============================================================ # 用户分页查询 # ============================================================ -@user_blue.route('/page', methods=["GET"]) + +@user_blue.route("/page", methods=["GET"]) @check_permission("sys:user:page") def page(): """ @@ -82,15 +85,15 @@ def page(): Returns: JSON: 用户分页数据，包含 records（数据列表）、total（总记录数）等 """ - result = user.get_user_page() - return result + return R.ok(data=user.get_user_page(request.args)) # ============================================================ # 用户详情查询 # ============================================================ -@user_blue.route('/detail/', methods=["GET"]) + +@user_blue.route("/detail/", methods=["GET"]) @check_permission("sys:user:detail") def detail(id): """ @@ -110,7 +113,8 @@ def detail(id): # 用户添加 # ============================================================ -@user_blue.route('/add', methods=["POST"]) + +@user_blue.route("/add", methods=["POST"]) @check_permission("sys:user:add") @check_demo def add(): @@ -123,15 +127,19 @@ def add(): Returns: JSON: 操作结果 """ - result = user.add_user() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + user.add_user(json_data) + return R.ok(msg="添加成功") # ============================================================ # 用户更新 # ============================================================ -@user_blue.route('/update', methods=["PUT"]) + +@user_blue.route("/update", methods=["PUT"]) @check_permission("sys:user:update") @check_demo def update(): @@ -144,15 +152,19 @@ def update(): Returns: JSON: 操作结果 """ - result = user.update_user() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + user.update_user(json_data) + return R.ok(msg="更新成功") # ============================================================ # 用户删除 # ============================================================ -@user_blue.route('/delete/', methods=["DELETE"]) + +@user_blue.route("/delete/", methods=["DELETE"]) @check_permission("sys:user:delete") @check_demo def delete(id): @@ -165,15 +177,16 @@ def delete(id): Returns: JSON: 操作结果 """ - result = user.delete_users(id) - return result + count = user.delete_users(id) + return R.ok(msg=f"本次共删除{count}条数据") # ============================================================ # 用户批量删除 # ============================================================ -@user_blue.route('/batchDelete', methods=["DELETE"]) + +@user_blue.route("/batchDelete", methods=["DELETE"]) @check_permission("sys:user:batch_delete") @check_demo def batch_delete(): @@ -192,14 +205,16 @@ def batch_delete(): id_str, err = parse_batch_ids() if err: return err - return user.delete_users(id_str) + count = user.delete_users(id_str) + return R.ok(msg=f"本次共删除{count}条数据") # ============================================================ # 用户状态设置 # ============================================================ -@user_blue.route('/status', methods=["PUT"]) + +@user_blue.route("/status", methods=["PUT"]) @check_permission("sys:user:status") @check_demo def status(): @@ -212,15 +227,19 @@ def status(): Returns: JSON: 操作结果 """ - result = user.update_user_status() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + user.update_user_status(json_data) + return R.ok(msg="更新成功") # ============================================================ # 用户密码重置 # ============================================================ -@user_blue.route('/resetPwd', methods=["PUT"]) + +@user_blue.route("/resetPwd", methods=["PUT"]) @check_permission("sys:user:resetPwd") @check_demo def reset_pwd(): diff --git a/src/modules/system/user/service.py b/src/modules/system/user/service.py index 239d0de..ad01347 100644 --- a/src/modules/system/user/service.py +++ b/src/modules/system/user/service.py @@ -64,8 +64,6 @@ import logging from datetime import datetime, timezone -from flask import request -from sqlalchemy import and_ from src.common.constants.dictionary_constants import GENDER_LIST from src.modules.system.user.schema import UserForm, UserStatusForm, UpdatePwdForm, UserInfoForm, ResetPwdForm @@ -73,19 +71,20 @@ from src.modules.system.dept.models import Dept from src.modules.system.level.models import Level from src.modules.system.position.models import Position from src.modules.system.user.models import User +from src.modules.system.user.repository import UserRepository +from src.core.exceptions import BizError from src.modules.association.user_role.models import UserRole from src.modules.association.user_role import service as user_role from src.modules.system.menu.service import invalidate_user_cache from src.common.constants.common import DEFAULT_PASSWORD, PAGE_SIZE from src.core.database import db -from src.core import response as R from src.common.utils import regular from src.core import password from src.common.utils.case_converter import convert_camel_to_snake from src.common.utils.rich_text import get_file_url, save_file from src.common.utils.model_utils import models_to_list, model_to_dict from src.core.password import verify_password, encrypt_password, generate_salt -from src.common.utils.request import parse_request_body, parse_id_list, like_contains +from src.common.utils.request import parse_id_list, like_contains from src.core.security import get_username, get_user_id # 获取日志记录器 @@ -96,7 +95,8 @@ logger = logging.getLogger(__name__) # 用户分页查询 # ============================================================ -def get_user_page(): + +def get_user_page(params: dict) -> dict: """ 查询用户分页数据 @@ -118,31 +118,26 @@ def get_user_page(): "pages": 总页数 } """ - try: - # 获取分页参数 - page_no = int(request.args.get('page', 1)) - page_size = int(request.args.get('limit', PAGE_SIZE)) + # 获取分页参数 + page_no = int(params.get("page", 1)) + page_size = int(params.get("limit", PAGE_SIZE)) - # 应用筛选条件 - query = _apply_filters() + # 应用筛选条件 + query = _apply_filters(params) - # 记录总数 - total = query.count() - # 计算总页数 - pages = (total + page_size - 1) // page_size if total > 0 else 0 + # 记录总数 + total = query.count() + # 计算总页数 + pages = (total + page_size - 1) // page_size if total > 0 else 0 - # 分页查询 - user_list = query.limit(page_size).offset((page_no - 1) * page_size).all() - # 构建返回数据 - page_data = _build_page_data(user_list, total, page_size, page_no, pages) - return R.ok(data=page_data) - - except Exception as e: - logger.error(f"查询用户分页数据异常: {str(e)}") - return R.failed(msg="查询失败，请稍后重试") + # 分页查询 + user_list = query.limit(page_size).offset((page_no - 1) * page_size).all() + # 构建返回数据 + page_data = _build_page_data(user_list, total, page_size, page_no, pages) + return page_data -def _apply_filters(): +def _apply_filters(params: dict): """ 应用查询筛选条件 @@ -160,17 +155,17 @@ def _apply_filters(): query = User.query.filter(User.is_delete == 0) # 用户姓名模糊筛选 - realname = request.args.get('realname') + realname = params.get("realname") if realname: - query = query.filter(User.realname.like(like_contains(realname), escape='\\')) + query = query.filter(User.realname.like(like_contains(realname), escape="\\")) # 性别精确筛选 - gender = request.args.get('gender') + gender = params.get("gender") if gender: query = query.filter(User.gender == gender) # 用户状态精确筛选 - status = request.args.get('status') + status = params.get("status") if status: query = query.filter(User.status == status) @@ -197,13 +192,7 @@ def _build_page_data(user_list, total, page_size, page_no, pages): # 补充关联数据 records = _enrich_user_records(records) - return { - 'records': records, - 'total': total, - 'size': page_size, - 'current': page_no, - 'pages': pages - } + return {"records": records, "total": total, "size": page_size, "current": page_no, "pages": pages} def _enrich_user_records(records): @@ -225,27 +214,27 @@ def _enrich_user_records(records): position_map = _get_position_map() # 批量查询所有用户的角色关系（1次查询替代N次） - user_ids = [data.get('id') for data in records] + user_ids = [data.get("id") for data in records] role_map = user_role.get_user_roles_batch(user_ids) for data in records: # 城市编码列表 - city_list = [data.get('provinceCode'), data.get('cityCode'), data.get('districtCode'), data.get('streetCode')] + city_list = [data.get("provinceCode"), data.get("cityCode"), data.get("districtCode"), data.get("streetCode")] # 性别名称 - data['genderName'] = GENDER_LIST.get(data.get('gender')) + data["genderName"] = GENDER_LIST.get(data.get("gender")) # 头像URL - data['avatar'] = get_file_url(data.get('avatar')) + data["avatar"] = get_file_url(data.get("avatar")) # 部门名称 - data['deptName'] = dept_map.get(data.get('deptId')) + data["deptName"] = dept_map.get(data.get("deptId")) # 职级名称 - data['levelName'] = level_map.get(data.get('levelId')) + data["levelName"] = level_map.get(data.get("levelId")) # 岗位名称 - data['positionName'] = position_map.get(data.get('positionId')) + data["positionName"] = position_map.get(data.get("positionId")) # 用户角色列表 - data['roles'] = role_map.get(data.get('id'), []) + data["roles"] = role_map.get(data.get("id"), []) # 行政区划 - data['city'] = city_list + data["city"] = city_list return records @@ -272,6 +261,7 @@ def _get_position_map(): # 用户详情查询 # ============================================================ + def get_user_detail(user_id): """ 根据ID查询用户详情（返回字典格式） @@ -295,7 +285,7 @@ def get_user_detail(user_id): # 移除密码字段 if data: - data.pop('password', None) + data.pop("password", None) # 补充关联数据 data = _enrich_user_detail(data) @@ -320,15 +310,15 @@ def _enrich_user_detail(data): return data # 头像URL - data['avatar'] = get_file_url(data.get('avatar')) + data["avatar"] = get_file_url(data.get("avatar")) # 获取用户角色数据（复用 user_role 服务） - role_list = user_role.get_user_role_list(data.get('id')) - data['roles'] = [role['id'] for role in role_list] + role_list = user_role.get_user_role_list(data.get("id")) + data["roles"] = [role["id"] for role in role_list] # 行政区划 - city_list = [data.get('provinceCode'), data.get('cityCode'), data.get('districtCode'), data.get('streetCode')] - data['city'] = city_list + city_list = [data.get("provinceCode"), data.get("cityCode"), data.get("districtCode"), data.get("streetCode")] + data["city"] = city_list return data @@ -337,6 +327,7 @@ def _enrich_user_detail(data): # 用户信息查询（根据ID）- 核心查询方法 # ============================================================ + def get_info(user_id: int): """ 根据ID查询用户信息（返回模型对象） @@ -362,27 +353,10 @@ def get_info(user_id: int): if user: print(user.realname) """ - # 参数有效性验证 - if not user_id or user_id <= 0: - return None - - try: - # 查询未删除的用户 - user = User.query.filter( - and_(User.id == user_id, User.is_delete == 0) - ).first() - - return user - except Exception as e: - logger.error(f"查询用户信息异常，ID: {user_id}, 错误: {str(e)}") - return None + return UserRepository.get_by_id(user_id) -# ============================================================ -# 用户添加 -# ============================================================ - -def add_user(): +def add_user(data: dict) -> None: """ 添加用户 @@ -396,69 +370,58 @@ def add_user(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = UserForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = UserForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 图片处理 - avatar = form.data.get('avatar') - if avatar: - form.avatar.data = save_file(avatar, "user") + # 图片处理 + avatar = form.data.get("avatar") + if avatar: + form.avatar.data = save_file(avatar, "user") - # 密码加密（未设置密码时使用默认密码） - pwd = form.data.get('password') or DEFAULT_PASSWORD - salt = password.generate_salt() - form.password.data = encrypt_password(pwd, salt) - form.salt.data = salt + # 密码加密（未设置密码时使用默认密码） + pwd = form.data.get("password") or DEFAULT_PASSWORD + salt = password.generate_salt() + form.password.data = encrypt_password(pwd, salt) + form.salt.data = salt - # 处理行政区划 - city = form.data.get('city', []) - if len(city) != 4: - return R.failed("请选择完整的所在城市（省/市/区/街道）") - form.provinceCode.data = city[0] - form.cityCode.data = city[1] - form.districtCode.data = city[2] - form.streetCode.data = city[3] + # 处理行政区划 + city = form.data.get("city", []) + if len(city) != 4: + raise BizError("请选择完整的所在城市（省/市/区/街道）") + form.provinceCode.data = city[0] + form.cityCode.data = city[1] + form.districtCode.data = city[2] + form.streetCode.data = city[3] - # 用户角色 - roles = form.data.get('roles') + # 用户角色 + roles = form.data.get("roles") - # 移除不需要直接保存到User表的字段 - form_data = form.data.copy() - form_data.pop('city', None) - form_data.pop('roles', None) + # 移除不需要直接保存到User表的字段 + form_data = form.data.copy() + form_data.pop("city", None) + form_data.pop("roles", None) - # 创建用户 - user = User(**form_data) - user.create_user = get_username() - user.save() + # 创建用户 + user = User(**form_data) + user.create_user = get_username() + user.save() - # 创建用户角色数据 - _add_user_role(user.id, roles) + # 创建用户角色数据 + _add_user_role(user.id, roles) - # 清除该用户权限缓存 - invalidate_user_cache(user.id) - - return R.ok(msg="添加成功") - - except Exception as e: - logger.error(f"添加用户异常: {str(e)}") - return R.failed(msg="添加用户失败，请稍后重试") + # 清除该用户权限缓存 + invalidate_user_cache(user.id) # ============================================================ # 用户更新 # ============================================================ -def update_user(): + +def update_user(data: dict) -> None: """ 更新用户 @@ -473,83 +436,72 @@ def update_user(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = UserForm(data=json_data) - if not form.validate(): - return R.failed(msg=regular.get_err(form)) + # 表单验证 + form = UserForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 获取用户ID - user_id = form.data.get('id') - if not user_id or int(user_id) <= 0: - return R.failed("用户ID无效") - user_id = int(user_id) + # 获取用户ID + user_id = form.data.get("id") + if not user_id or int(user_id) <= 0: + raise BizError("用户ID无效") + user_id = int(user_id) - # 复用 get_info 方法验证用户是否存在 - user = get_info(user_id) - if not user: - return R.failed("用户不存在") + # 复用 get_info 方法验证用户是否存在 + user = get_info(user_id) + if not user: + raise BizError("用户不存在") - # 图片处理 - avatar = form.data.get('avatar') - if avatar: - form.avatar.data = save_file(avatar, "user") + # 图片处理 + avatar = form.data.get("avatar") + if avatar: + form.avatar.data = save_file(avatar, "user") - # 密码加密（仅当有传入新密码时） - pwd = form.data.get('password') - if pwd: - salt = password.generate_salt() - form.password.data = encrypt_password(pwd, salt) - form.salt.data = salt - else: - # 密码不填时保持原密码值不做更新 - form_data = form.data.copy() - form_data.pop('password', None) - form_data.pop('salt', None) + # 密码加密（仅当有传入新密码时） + pwd = form.data.get("password") + if pwd: + salt = password.generate_salt() + form.password.data = encrypt_password(pwd, salt) + form.salt.data = salt + else: + # 密码不填时保持原密码值不做更新 + form_data = form.data.copy() + form_data.pop("password", None) + form_data.pop("salt", None) - # 处理行政区划 - city = form.data.get('city', []) - if len(city) != 4: - return R.failed("请选择完整的所在城市（省/市/区/街道）") - form.provinceCode.data = city[0] - form.cityCode.data = city[1] - form.districtCode.data = city[2] - form.streetCode.data = city[3] + # 处理行政区划 + city = form.data.get("city", []) + if len(city) != 4: + raise BizError("请选择完整的所在城市（省/市/区/街道）") + form.provinceCode.data = city[0] + form.cityCode.data = city[1] + form.districtCode.data = city[2] + form.streetCode.data = city[3] - # 用户角色 - roles = form.data.get('roles') + # 用户角色 + roles = form.data.get("roles") - # 移除不需要直接更新到User表的字段 - form_data = convert_camel_to_snake(form.data.copy()) - form_data.pop('id', None) - form_data.pop('city', None) - form_data.pop('roles', None) + # 移除不需要直接更新到User表的字段 + form_data = convert_camel_to_snake(form.data.copy()) + form_data.pop("id", None) + form_data.pop("city", None) + form_data.pop("roles", None) - # 更新用户信息 - form_data['update_time'] = datetime.now(tz=timezone.utc) - form_data['update_user'] = get_username() + # 更新用户信息 + form_data["update_time"] = datetime.now(tz=timezone.utc) + form_data["update_user"] = get_username() - result = User.query.filter_by(id=user_id).update(form_data) + result = User.query.filter_by(id=user_id).update(form_data) - if not result: - return R.failed("更新失败") + if not result: + raise BizError("更新失败") - # 更新用户角色数据 - _add_user_role(user_id, roles) + # 更新用户角色数据 + _add_user_role(user_id, roles) - # 清除该用户权限缓存 - invalidate_user_cache(user_id) - - return R.ok(msg="更新成功") - - except Exception as e: - logger.error(f"更新用户异常: {str(e)}") - return R.failed(msg="更新用户失败，请稍后重试") + # 清除该用户权限缓存 + invalidate_user_cache(user_id) def _add_user_role(user_id, roles): @@ -566,17 +518,15 @@ def _add_user_role(user_id, roles): # 批量插入 if roles: - db.session.bulk_insert_mappings( - UserRole, - [{'user_id': user_id, 'role_id': role_id} for role_id in roles] - ) + db.session.bulk_insert_mappings(UserRole, [{"user_id": user_id, "role_id": role_id} for role_id in roles]) # ============================================================ # 用户删除 # ============================================================ -def delete_users(user_ids): + +def delete_users(user_ids: str) -> int: """ 删除用户（批量删除） @@ -589,14 +539,18 @@ def delete_users(user_ids): Returns: R: 操作结果 """ - return User.batch_delete_r(user_ids) + user_ids_list = parse_id_list(user_ids) + if not user_ids_list: + raise BizError("无效的记录ID") + return User.batch_delete(user_ids_list) # ============================================================ # 用户状态设置 # ============================================================ -def update_user_status(): + +def update_user_status(data: dict) -> None: """ 设置用户状态 @@ -605,45 +559,34 @@ def update_user_status(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = UserStatusForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = UserStatusForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - user_id = int(form.data['id']) - status = int(form.data['status']) + user_id = int(form.data["id"]) + status = int(form.data["status"]) - # 复用 get_info 方法验证用户是否存在 - user = get_info(user_id) - if not user: - return R.failed("用户不存在") + # 复用 get_info 方法验证用户是否存在 + user = get_info(user_id) + if not user: + raise BizError("用户不存在") - # 更新状态 - update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} - result = User.query.filter_by(id=user_id).update(update_data) + # 更新状态 + update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} + result = User.query.filter_by(id=user_id).update(update_data) - if not result: - return R.failed("设置失败") - - return R.ok(msg="设置成功") - - except Exception as e: - logger.error(f"设置用户状态异常: {str(e)}") - return R.failed(msg="设置用户状态失败，请稍后重试") + if not result: + raise BizError("设置失败") # ============================================================ # 用户信息更新（个人信息） # ============================================================ -def update_user_info(): + +def update_user_info(data: dict) -> None: """ 更新当前用户信息 @@ -652,54 +595,43 @@ def update_user_info(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = UserInfoForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = UserInfoForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 用户ID（当前登录用户） - user_id = get_user_id() + # 用户ID（当前登录用户） + user_id = get_user_id() - # 复用 get_info 方法验证用户是否存在 - user = get_info(user_id) - if not user: - return R.failed("用户不存在") + # 复用 get_info 方法验证用户是否存在 + user = get_info(user_id) + if not user: + raise BizError("用户不存在") - # 更新用户信息 - update_data = { - 'realname': form.realname.data, - 'gender': form.gender.data, - 'mobile': form.mobile.data, - 'email': form.email.data, - 'address': form.address.data, - 'intro': form.intro.data, - 'update_time': datetime.now(tz=timezone.utc), - 'update_user': get_username() - } - result = User.query.filter_by(id=user_id).update(update_data) + # 更新用户信息 + update_data = { + "realname": form.realname.data, + "gender": form.gender.data, + "mobile": form.mobile.data, + "email": form.email.data, + "address": form.address.data, + "intro": form.intro.data, + "update_time": datetime.now(tz=timezone.utc), + "update_user": get_username(), + } + result = User.query.filter_by(id=user_id).update(update_data) - if not result: - return R.failed("更新失败") - - return R.ok(msg="更新成功") - - except Exception as e: - logger.error(f"更新用户信息异常: {str(e)}") - return R.failed(msg="更新用户信息失败，请稍后重试") + if not result: + raise BizError("更新失败") # ============================================================ # 密码修改 # ============================================================ -def update_password(): + +def update_password(data: dict) -> None: """ 修改当前用户密码 @@ -708,51 +640,45 @@ def update_password(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = UpdatePwdForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = UpdatePwdForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 用户ID（当前登录用户） - user_id = get_user_id() + # 用户ID（当前登录用户） + user_id = get_user_id() - # 根据ID查询用户 - user = get_info(user_id) - if not user: - return R.failed("用户不存在") + # 根据ID查询用户 + user = get_info(user_id) + if not user: + raise BizError("用户不存在") - # 验证原始密码 - if not verify_password(form.oldPassword.data, user.salt, user.password): - return R.failed("原始密码不正确") + # 验证原始密码 + if not verify_password(form.oldPassword.data, user.salt, user.password): + raise BizError("原始密码不正确") - # 加密新密码 - new_salt = generate_salt() - password = encrypt_password(form.newPassword.data, new_salt) - update_data = {"password": password, "salt": new_salt, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} - result = User.query.filter_by(id=user_id).update(update_data) + # 加密新密码 + new_salt = generate_salt() + password = encrypt_password(form.newPassword.data, new_salt) + update_data = { + "password": password, + "salt": new_salt, + "update_time": datetime.now(tz=timezone.utc), + "update_user": get_username(), + } + result = User.query.filter_by(id=user_id).update(update_data) - if not result: - return R.failed("更新失败") - - return R.ok(msg="更新成功") - - except Exception as e: - logger.error(f"修改密码异常: {str(e)}") - return R.failed(msg="修改密码失败，请稍后重试") + if not result: + raise BizError("更新失败") # ============================================================ # 密码重置 # ============================================================ -def reset_user_password(): + +def reset_user_password(data: dict) -> None: """ 重置用户密码 @@ -761,36 +687,29 @@ def reset_user_password(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = ResetPwdForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = ResetPwdForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - user_id = form.userId.data + user_id = form.userId.data - # 复用 get_info 方法验证用户是否存在 - user = get_info(user_id) - if not user: - return R.failed("用户不存在") + # 复用 get_info 方法验证用户是否存在 + user = get_info(user_id) + if not user: + raise BizError("用户不存在") - # 加密新密码（默认密码123456） - new_salt = generate_salt() - password = encrypt_password(DEFAULT_PASSWORD, new_salt) - update_data = {"password": password, "salt": new_salt, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} - result = User.query.filter_by(id=user_id).update(update_data) + # 加密新密码（默认密码123456） + new_salt = generate_salt() + password = encrypt_password(DEFAULT_PASSWORD, new_salt) + update_data = { + "password": password, + "salt": new_salt, + "update_time": datetime.now(tz=timezone.utc), + "update_user": get_username(), + } + result = User.query.filter_by(id=user_id).update(update_data) - if not result: - return R.failed("重置密码失败") - - return R.ok(msg="重置密码成功") - - except Exception as e: - logger.error(f"重置密码异常: {str(e)}") - return R.failed(msg="重置密码失败，请稍后重试") + if not result: + raise BizError("重置密码失败")