From 9acbf6275e9d0e721c74f4f01af1dea9f48443a4 Mon Sep 17 00:00:00 2001 From: DjangoAdmin Dev Date: Fri, 7 Aug 2026 16:26:33 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20P4c=20=E6=9C=80=E7=BB=88=E6=89=B9?= =?UTF-8?q?=E6=AC=A1-2=20dictionary=20=E6=A8=A1=E5=9D=97=EF=BC=88dict+dict?= =?UTF-8?q?=5Fitem=20=E5=90=88=E5=B9=B6=E6=9C=8D=E5=8A=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【修复合并遗留】 - 重命名重名函数（_apply_dict_item_filters/get_dict_item_info/ _build_dict_item_page_data），修复 get_info 调用点 - router 双 blueprint 合并导入、dict_item 视图函数重命名 - schema/service/models 中部重复 import 行级去重 - 修复 schema typo DictItemItem - models.py 误改后从 git 恢复（P4c 不需改 models） 【教训】regex 改复杂合并文件易误伤（.*?跨 docstring 匹配删了顶部导入）， 已恢复并改用精确行级去重 【验收】flake8 零告警；应用启动；dict/dict_item 分页运行时正常；12 passed Co-Authored-By: Claude --- .../dictionary/dict_item_repository.py | 15 + src/modules/dictionary/dict_repository.py | 15 + src/modules/dictionary/models.py | 70 +--- src/modules/dictionary/router.py | 116 +++--- src/modules/dictionary/schema.py | 111 ++--- src/modules/dictionary/service.py | 385 +++++++----------- src/modules/log/service.py | 2 +- 7 files changed, 288 insertions(+), 426 deletions(-) create mode 100644 src/modules/dictionary/dict_item_repository.py create mode 100644 src/modules/dictionary/dict_repository.py diff --git a/src/modules/dictionary/dict_item_repository.py b/src/modules/dictionary/dict_item_repository.py new file mode 100644 index 0000000..561c998 --- /dev/null +++ b/src/modules/dictionary/dict_item_repository.py @@ -0,0 +1,15 @@ +""" +DictItem 仓储模块 +============================================================= + +数据访问层，继承 BaseRepository 获得通用 CRUD 能力。 +""" + +from src.common.base_repository import BaseRepository +from src.modules.dictionary.models import DictItem + + +class DictItemRepository(BaseRepository): + """DictItem 仓储""" + + model = DictItem diff --git a/src/modules/dictionary/dict_repository.py b/src/modules/dictionary/dict_repository.py new file mode 100644 index 0000000..43fe680 --- /dev/null +++ b/src/modules/dictionary/dict_repository.py @@ -0,0 +1,15 @@ +""" +Dict 仓储模块 +============================================================= + +数据访问层，继承 BaseRepository 获得通用 CRUD 能力。 +""" + +from src.common.base_repository import BaseRepository +from src.modules.dictionary.models import Dict + + +class DictRepository(BaseRepository): + """Dict 仓储""" + + model = Dict diff --git a/src/modules/dictionary/models.py b/src/modules/dictionary/models.py index 1a9fa0e..a8f1fa5 100644 --- a/src/modules/dictionary/models.py +++ b/src/modules/dictionary/models.py @@ -50,13 +50,13 @@ 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 # ============================================================ # 字典模型 # ============================================================ + class Dict(BaseModel, BaseDB): """ 字典分组模型 @@ -84,32 +84,16 @@ class Dict(BaseModel, BaseDB): # ============================================================ # 字典名称 - name = db.Column( - db.String(150), - nullable=False, - comment="字典名称" - ) + name = db.Column(db.String(150), nullable=False, comment="字典名称") # 字典编码 - code = db.Column( - db.String(150), - nullable=False, - comment="字典编码" - ) + code = db.Column(db.String(150), nullable=False, comment="字典编码") # 字典排序 - sort = db.Column( - db.Integer, - default=0, - comment="字典排序" - ) + sort = db.Column(db.Integer, default=0, comment="字典排序") # 字典备注 - note = db.Column( - db.String(255), - nullable=True, - comment="字典备注" - ) + note = db.Column(db.String(255), nullable=True, comment="字典备注") # ============================================================ # 魔法方法 @@ -125,7 +109,6 @@ class Dict(BaseModel, BaseDB): return "字典{}".format(self.id) - """ 字典项模型模块 ============================================================= @@ -133,17 +116,12 @@ class Dict(BaseModel, BaseDB): 提供数据字典项的数据模型定义，用于管理字典的具体选项。 """ -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 - # ============================================================ # 字典项模型 # ============================================================ + class DictItem(BaseModel, BaseDB): """ 字典项模型 @@ -172,46 +150,22 @@ class DictItem(BaseModel, BaseDB): # ============================================================ # 字典ID - dict_id = db.Column( - db.Integer, - default=0, - comment="字典ID" - ) + dict_id = db.Column(db.Integer, default=0, comment="字典ID") # 字典编码 - dict_code = db.Column( - db.String(100), - nullable=False, - comment="字典编码" - ) + dict_code = db.Column(db.String(100), nullable=False, comment="字典编码") # 字典项名称 - name = db.Column( - db.String(100), - nullable=False, - comment="字典项名称" - ) + name = db.Column(db.String(100), nullable=False, comment="字典项名称") # 字典项值 - value = db.Column( - db.String(100), - nullable=False, - comment="字典项值" - ) + value = db.Column(db.String(100), nullable=False, comment="字典项值") # 字典项排序 - sort = db.Column( - db.Integer, - default=0, - comment="字典项顺序" - ) + sort = db.Column(db.Integer, default=0, comment="字典项顺序") # 字典项备注 - note = db.Column( - db.String(255), - nullable=True, - comment="字典项备注" - ) + note = db.Column(db.String(255), nullable=True, comment="字典项备注") # ============================================================ # 魔法方法 diff --git a/src/modules/dictionary/router.py b/src/modules/dictionary/router.py index 50e7952..42826a2 100644 --- a/src/modules/dictionary/router.py +++ b/src/modules/dictionary/router.py @@ -53,22 +53,25 @@ 依赖模块： application.services.dicts - 字典业务逻辑层 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.dictionary import service as dict -from src.common.utils.request import parse_batch_ids -dict_blue = Blueprint('dict', __name__, url_prefix='/dict') +from src.modules.dictionary import service as dict_item +from src.common.utils.request import parse_batch_ids, parse_request_body from src.core import response as R +dict_blue = Blueprint("dict", __name__, url_prefix="/dict") + # ============================================================ # 字典分页查询 # ============================================================ -@dict_blue.route('/page', methods=["GET"]) + +@dict_blue.route("/page", methods=["GET"]) @check_permission("sys:dict:page") def page(): """ @@ -80,15 +83,15 @@ def page(): Returns: JSON: 字典分页数据，包含 records（数据列表）、total（总记录数）等 """ - result = dict.get_dict_page() - return result + return R.ok(data=dict.get_dict_page(request.args)) # ============================================================ # 字典详情查询 # ============================================================ -@dict_blue.route('/detail/', methods=["GET"]) + +@dict_blue.route("/detail/", methods=["GET"]) @check_permission("sys:dict:detail") def detail(id): """ @@ -108,7 +111,8 @@ def detail(id): # 字典添加 # ============================================================ -@dict_blue.route('/add', methods=["POST"]) + +@dict_blue.route("/add", methods=["POST"]) @check_permission("sys:dict:add") @check_demo def add(): @@ -121,15 +125,19 @@ def add(): Returns: JSON: 操作结果 """ - result = dict.add_dict() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + dict.add_dict(json_data) + return R.ok(msg="添加成功") # ============================================================ # 字典更新 # ============================================================ -@dict_blue.route('/update', methods=["PUT"]) + +@dict_blue.route("/update", methods=["PUT"]) @check_permission("sys:dict:update") @check_demo def update(): @@ -142,15 +150,19 @@ def update(): Returns: JSON: 操作结果 """ - result = dict.update_dict() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + dict.update_dict(json_data) + return R.ok(msg="更新成功") # ============================================================ # 字典删除 # ============================================================ -@dict_blue.route('/delete/', methods=["DELETE"]) + +@dict_blue.route("/delete/", methods=["DELETE"]) @check_permission("sys:dict:delete") @check_demo def delete(id): @@ -163,15 +175,16 @@ def delete(id): Returns: JSON: 操作结果 """ - result = dict.delete_dicts(id) - return result + count = dict.delete_dicts(id) + return R.ok(msg=f"本次共删除{count}条数据") # ============================================================ # 字典批量删除 # ============================================================ -@dict_blue.route('/batchDelete', methods=["DELETE"]) + +@dict_blue.route("/batchDelete", methods=["DELETE"]) @check_permission("sys:dict:batch_delete") @check_demo def batch_delete(): @@ -190,9 +203,8 @@ def batch_delete(): id_str, err = parse_batch_ids() if err: return err - return dict.delete_dicts(id_str) - - + count = dict.delete_dicts(id_str) + return R.ok(msg=f"本次共删除{count}条数据") """ @@ -208,24 +220,17 @@ def batch_delete(): 依赖模块： application.services.dicts_data - 字典项业务逻辑层 application.middleware.check_permission - 权限校验中间件 """ -from flask import Blueprint - -from application.middleware.check_demo import check_demo -from application.middleware.check_permission import check_permission -from src.modules.dictionary import service as dict_item -from src.common.utils.request import parse_batch_ids -dict_item_blue = Blueprint('dict_item', __name__, url_prefix='/dict/item') - -from src.core import response as R +dict_item_blue = Blueprint("dict_item", __name__, url_prefix="/dict/item") # ============================================================ # 字典项分页查询 # ============================================================ -@dict_item_blue.route('/page', methods=["GET"]) + +@dict_item_blue.route("/page", methods=["GET"]) @check_permission("sys:dictitem:page") -def page(): +def dict_item_page(): """ 查询字典项分页数据 @@ -235,17 +240,17 @@ def page(): Returns: JSON: 字典项分页数据，包含 records（数据列表）、total（总记录数）等 """ - result = dict_item.get_dict_item_page() - return result + return R.ok(data=dict_item.get_dict_item_page(request.args)) # ============================================================ # 字典项详情查询 # ============================================================ -@dict_item_blue.route('/detail/', methods=["GET"]) + +@dict_item_blue.route("/detail/", methods=["GET"]) @check_permission("sys:dictitem:detail") -def detail(id): +def dict_item_detail(id): """ 查询字典项详情 @@ -263,10 +268,11 @@ def detail(id): # 字典项添加 # ============================================================ -@dict_item_blue.route('/add', methods=["POST"]) + +@dict_item_blue.route("/add", methods=["POST"]) @check_permission("sys:dictitem:add") @check_demo -def add(): +def dict_item_add(): """ 添加字典项 @@ -276,18 +282,22 @@ def add(): Returns: JSON: 操作结果 """ - result = dict_item.add_dict_item() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + dict_item.add_dict_item(json_data) + return R.ok(msg="添加成功") # ============================================================ # 字典项更新 # ============================================================ -@dict_item_blue.route('/update', methods=["PUT"]) + +@dict_item_blue.route("/update", methods=["PUT"]) @check_permission("sys:dictitem:update") @check_demo -def update(): +def dict_item_update(): """ 更新字典项 @@ -297,18 +307,22 @@ def update(): Returns: JSON: 操作结果 """ - result = dict_item.update_dict_item() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + dict_item.update_dict_item(json_data) + return R.ok(msg="更新成功") # ============================================================ # 字典项删除 # ============================================================ -@dict_item_blue.route('/delete/', methods=["DELETE"]) + +@dict_item_blue.route("/delete/", methods=["DELETE"]) @check_permission("sys:dictitem:delete") @check_demo -def delete(id): +def dict_item_delete(id): """ 删除字典项 @@ -318,18 +332,19 @@ def delete(id): Returns: JSON: 操作结果 """ - result = dict_item.delete_dict_items(id) - return result + count = dict_item.delete_dict_items(id) + return R.ok(msg=f"本次共删除{count}条数据") # ============================================================ # 字典项批量删除 # ============================================================ -@dict_item_blue.route('/batchDelete', methods=["DELETE"]) + +@dict_item_blue.route("/batchDelete", methods=["DELETE"]) @check_permission("sys:dictitem:batch_delete") @check_demo -def batch_delete(): +def dict_item_batch_delete(): """ 批量删除字典项 @@ -345,4 +360,5 @@ def batch_delete(): id_str, err = parse_batch_ids() if err: return err - return dict_item.delete_dict_items(id_str) + count = dict_item.delete_dict_items(id_str) + return R.ok(msg=f"本次共删除{count}条数据") diff --git a/src/modules/dictionary/schema.py b/src/modules/dictionary/schema.py index 5a9cfb9..b350de7 100644 --- a/src/modules/dictionary/schema.py +++ b/src/modules/dictionary/schema.py @@ -51,13 +51,14 @@ from wtforms import ValidationError, IntegerField, StringField from wtforms.validators import NumberRange, DataRequired, Length, Optional from src.common.forms.base_form import BaseForm -from src.modules.dictionary.models import Dict +from src.modules.dictionary.models import Dict, DictItem # ============================================================ # 字典表单验证（添加和更新共用） # ============================================================ + class DictForm(BaseForm): """ 字典表单验证类 @@ -77,59 +78,45 @@ class DictForm(BaseForm): # 字典ID # ============================================================ id = IntegerField( - label='字典ID', - validators=[ - Optional(), # 允许为空（添加时） - NumberRange(min=1, message='字典ID必须大于0') - ], - description='字典唯一标识，更新时必填，添加时自动生成' + label="字典ID", + validators=[Optional(), NumberRange(min=1, message="字典ID必须大于0")], # 允许为空（添加时） + description="字典唯一标识，更新时必填，添加时自动生成", ) # ============================================================ # 字典名称 # ============================================================ name = StringField( - label='字典名称', - validators=[ - DataRequired(message='字典名称不能为空'), - Length(max=150, message='字典名称长度不得超过150个字符') - ], - description='字典显示名称，如：性别、状态' + label="字典名称", + validators=[DataRequired(message="字典名称不能为空"), Length(max=150, message="字典名称长度不得超过150个字符")], + description="字典显示名称，如：性别、状态", ) # ============================================================ # 字典编码 # ============================================================ code = StringField( - label='字典编码', - validators=[ - DataRequired(message='字典编码不能为空'), - Length(max=150, message='字典编码长度不得超过150个字符') - ], - description='字典唯一标识编码，如：gender、status' + label="字典编码", + validators=[DataRequired(message="字典编码不能为空"), Length(max=150, message="字典编码长度不得超过150个字符")], + description="字典唯一标识编码，如：gender、status", ) # ============================================================ # 字典排序 # ============================================================ sort = IntegerField( - label='字典排序', - validators=[ - Optional(), - NumberRange(min=0, max=99999, message='字典排序值在0~99999之间') - ], - description='排序值，数值越小越靠前' + label="字典排序", + validators=[Optional(), NumberRange(min=0, max=99999, message="字典排序值在0~99999之间")], + description="排序值，数值越小越靠前", ) # ============================================================ # 字典备注 # ============================================================ note = StringField( - label='字典备注', - validators=[ - Length(max=255, message='字典备注长度不得超过255个字符') - ], - description='字典补充说明' + label="字典备注", + validators=[Length(max=255, message="字典备注长度不得超过255个字符")], + description="字典补充说明", ) # ============================================================ @@ -199,7 +186,6 @@ class DictForm(BaseForm): raise ValidationError("字典编码已存在") - """ 字典项表单验证模块 ============================================================= @@ -208,17 +194,12 @@ class DictForm(BaseForm): - 字典项添加/更新表单验证 """ -from wtforms import ValidationError, IntegerField, StringField -from wtforms.validators import NumberRange, DataRequired, Length, Optional - -from src.common.forms.base_form import BaseForm -from src.modules.dictionary.models import DictItem - # ============================================================ # 字典项表单验证（添加和更新共用） # ============================================================ + class DictItemForm(BaseForm): """ 字典项表单验证类 @@ -239,71 +220,57 @@ class DictItemForm(BaseForm): # 字典项ID # ============================================================ id = IntegerField( - label='字典项ID', - validators=[ - Optional(), # 允许为空（添加时） - NumberRange(min=1, message='字典项ID必须大于0') - ], - description='字典项唯一标识，更新时必填，添加时自动生成' + label="字典项ID", + validators=[Optional(), NumberRange(min=1, message="字典项ID必须大于0")], # 允许为空（添加时） + description="字典项唯一标识，更新时必填，添加时自动生成", ) # ============================================================ # 字典项名称 # ============================================================ name = StringField( - label='字典项名称', + label="字典项名称", validators=[ - DataRequired(message='字典项名称不能为空'), - Length(max=150, message='字典项名称长度不得超过150个字符') + DataRequired(message="字典项名称不能为空"), + Length(max=150, message="字典项名称长度不得超过150个字符"), ], - description='字典项显示名称，如：男、女' + description="字典项显示名称，如：男、女", ) # ============================================================ # 字典项值 # ============================================================ value = StringField( - label='字典项值', - validators=[ - DataRequired(message='字典项值不能为空'), - Length(max=150, message='字典项值长度不得超过150个字符') - ], - description='字典项实际存储值，如：1、2' + label="字典项值", + validators=[DataRequired(message="字典项值不能为空"), Length(max=150, message="字典项值长度不得超过150个字符")], + description="字典项实际存储值，如：1、2", ) # ============================================================ # 字典ID # ============================================================ dictId = IntegerField( - label='字典ID', - validators=[ - DataRequired(message='字典ID不能为空'), - NumberRange(min=0, message='字典ID值不得小于0') - ], - description='所属字典分组ID' + label="字典ID", + validators=[DataRequired(message="字典ID不能为空"), NumberRange(min=0, message="字典ID值不得小于0")], + description="所属字典分组ID", ) # ============================================================ # 字典项备注 # ============================================================ note = StringField( - label='字典项备注', - validators=[ - Length(max=255, message='字典项备注长度不得超过255个字符') - ], - description='字典项补充说明' + label="字典项备注", + validators=[Length(max=255, message="字典项备注长度不得超过255个字符")], + description="字典项补充说明", ) # ============================================================ # 字典项排序 # ============================================================ sort = IntegerField( - label='字典项排序', - validators=[ - Optional(), - NumberRange(min=0, max=99999, message='字典项排序值在0~99999之间') - ], - description='排序值，数值越小越靠前' + label="字典项排序", + validators=[Optional(), NumberRange(min=0, max=99999, message="字典项排序值在0~99999之间")], + description="排序值，数值越小越靠前", ) # ============================================================ @@ -369,9 +336,7 @@ class DictItemForm(BaseForm): # 构建查询：查找同一字典下相同值的记录 query = DictItem.query.filter( - DictItem.dict_id == dict_id, - DictItem.value == field.data, - DictItem.is_delete == 0 + DictItem.dict_id == dict_id, DictItem.value == field.data, DictItem.is_delete == 0 ) # 如果是更新操作，排除当前记录自身 diff --git a/src/modules/dictionary/service.py b/src/modules/dictionary/service.py index 8ef95ee..51c4543 100644 --- a/src/modules/dictionary/service.py +++ b/src/modules/dictionary/service.py @@ -63,18 +63,17 @@ import logging from datetime import datetime, timezone -from flask import request -from sqlalchemy import and_ from src.common.constants.common import PAGE_SIZE -from src.modules.dictionary.schema import DictForm -from src.modules.dictionary.models import Dict -from src.core.database import db -from src.core import response as R +from src.modules.dictionary.schema import DictForm, DictItemForm +from src.modules.dictionary.models import Dict, DictItem +from src.modules.dictionary.dict_repository import DictRepository +from src.modules.dictionary.dict_item_repository import DictItemRepository +from src.core.exceptions import BizError from src.common.utils import regular from src.common.utils.case_converter import convert_camel_to_snake from src.common.utils.model_utils import models_to_list, model_to_dict -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 # 获取日志记录器 @@ -85,7 +84,8 @@ logger = logging.getLogger(__name__) # 字典类型分页查询 # ============================================================ -def get_dict_page(): + +def get_dict_page(params: dict) -> dict: """ 查询字典类型分页数据 @@ -106,31 +106,26 @@ def get_dict_page(): "pages": 总页数 } """ - try: - # 获取分页参数 - page_no = int(request.args.get('pageNo', 1)) - page_size = int(request.args.get('pageSize', PAGE_SIZE)) + # 获取分页参数 + page_no = int(params.get("pageNo", 1)) + page_size = int(params.get("pageSize", 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 - # 分页查询 - dict_list = query.limit(page_size).offset((page_no - 1) * page_size).all() - # 构建返回数据 - page_data = _build_page_data(dict_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="查询失败，请稍后重试") + # 分页查询 + dict_list = query.limit(page_size).offset((page_no - 1) * page_size).all() + # 构建返回数据 + page_data = _build_page_data(dict_list, total, page_size, page_no, pages) + return page_data -def _apply_filters(): +def _apply_filters(params: dict): """ 应用查询筛选条件 @@ -147,12 +142,12 @@ def _apply_filters(): query = Dict.query.filter(Dict.is_delete == 0) # 字典名称模糊筛选 - name = request.args.get('name') + name = params.get("name") if name: - query = query.filter(Dict.name.like(like_contains(name), escape='\\')) + query = query.filter(Dict.name.like(like_contains(name), escape="\\")) # 字典状态精确筛选 - status = request.args.get('status') + status = params.get("status") if status: query = query.filter(Dict.status == status) @@ -177,19 +172,14 @@ def _build_page_data(dict_list, total, page_size, page_no, pages): # 将模型对象列表转换为字典列表 records = models_to_list(dict_list) - 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 get_dict_detail(dict_id): """ 根据ID查询字典类型详情（返回字典格式） @@ -220,6 +210,7 @@ def get_dict_detail(dict_id): # 字典类型信息查询（根据ID）- 核心查询方法 # ============================================================ + def get_info(dict_id: int): """ 根据ID查询字典类型信息（返回模型对象） @@ -244,26 +235,10 @@ def get_info(dict_id: int): if dict_obj: print(dict_obj.name) """ - # 参数有效性验证 - if not dict_id or dict_id <= 0: - return None - - try: - # 查询未删除的字典类型 - dict_obj = Dict.query.filter( - and_(Dict.id == dict_id, Dict.is_delete == 0) - ).first() - return dict_obj - except Exception as e: - logger.error(f"查询字典类型信息异常，ID: {dict_id}, 错误: {str(e)}") - return None + return DictRepository.get_by_id(dict_id) -# ============================================================ -# 字典类型添加 -# ============================================================ - -def add_dict(): +def add_dict(data: dict) -> None: """ 添加字典类型 @@ -275,35 +250,24 @@ def add_dict(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = DictForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = DictForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 创建字典类型 - dict_obj = Dict(**form.data) - dict_obj.create_user = get_username() - dict_obj.save() - - return R.ok(msg="添加成功") - - except Exception as e: - logger.error(f"添加字典类型异常: {str(e)}") - return R.failed(msg="添加失败，请稍后重试") + # 创建字典类型 + dict_obj = Dict(**form.data) + dict_obj.create_user = get_username() + dict_obj.save() # ============================================================ # 字典类型更新 # ============================================================ -def update_dict(): + +def update_dict(data: dict) -> None: """ 更新字典类型 @@ -316,52 +280,42 @@ def update_dict(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = DictForm(data=json_data) - if not form.validate(): - return R.failed(msg=regular.get_err(form)) + # 表单验证 + form = DictForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 获取字典类型ID - dict_id = form.data.get('id') - if not dict_id or int(dict_id) <= 0: - return R.failed("字典类型ID无效") - dict_id = int(dict_id) + # 获取字典类型ID + dict_id = form.data.get("id") + if not dict_id or int(dict_id) <= 0: + raise BizError("字典类型ID无效") + dict_id = int(dict_id) - # 复用 get_info 方法验证字典类型是否存在 - dict_obj = get_info(dict_id) - if not dict_obj: - return R.failed("字典类型不存在") + # 复用 get_info 方法验证字典类型是否存在 + dict_obj = get_info(dict_id) + if not dict_obj: + raise BizError("字典类型不存在") - # 更新字典类型信息（移除ID，避免更新主键） - form_data = convert_camel_to_snake(form.data.copy()) - form_data.pop('id', None) + # 更新字典类型信息（移除ID，避免更新主键） + form_data = convert_camel_to_snake(form.data.copy()) + form_data.pop("id", 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 = Dict.query.filter_by(id=dict_id).update(form_data) + result = Dict.query.filter_by(id=dict_id).update(form_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 delete_dicts(dict_ids): + +def delete_dicts(dict_ids: str) -> int: """ 删除字典类型（批量删除） @@ -374,8 +328,10 @@ def delete_dicts(dict_ids): Returns: R: 操作结果 """ - return Dict.batch_delete_r(dict_ids) - + dict_ids_list = parse_id_list(dict_ids) + if not dict_ids_list: + raise BizError("无效的记录ID") + return Dict.batch_delete(dict_ids_list) """ @@ -399,23 +355,6 @@ def delete_dicts(dict_ids): 3. 数据维护：字典项的添加、更新、删除（支持批量） """ -import logging - -from datetime import datetime, timezone -from flask import request -from sqlalchemy import and_ - -from src.common.constants.common import PAGE_SIZE -from src.modules.dictionary.schema import DictItemForm -from src.modules.dictionary.models import Dict -from src.modules.dictionary.models import DictItem -from src.core.database import db -from src.core import response as R -from src.common.utils import regular -from src.common.utils.case_converter import convert_camel_to_snake -from src.common.utils.model_utils import models_to_list, model_to_dict -from src.common.utils.request import parse_request_body, parse_id_list, like_contains -from src.core.security import get_username # 获取日志记录器 logger = logging.getLogger(__name__) @@ -425,7 +364,8 @@ logger = logging.getLogger(__name__) # 字典项分页查询 # ============================================================ -def get_dict_item_page(): + +def get_dict_item_page(params: dict) -> dict: """ 查询字典项分页数据 @@ -447,31 +387,26 @@ def get_dict_item_page(): "pages": 总页数 } """ - try: - # 获取分页参数 - page_no = int(request.args.get('pageNo', 1)) - page_size = int(request.args.get('pageSize', PAGE_SIZE)) + # 获取分页参数 + page_no = int(params.get("pageNo", 1)) + page_size = int(params.get("pageSize", PAGE_SIZE)) - # 应用筛选条件 - query = _apply_filters() + # 应用筛选条件 + query = _apply_dict_item_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 - # 分页查询 - dict_item_list = query.limit(page_size).offset((page_no - 1) * page_size).all() - # 构建返回数据 - page_data = _build_page_data(dict_item_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="查询失败，请稍后重试") + # 分页查询 + dict_item_list = query.limit(page_size).offset((page_no - 1) * page_size).all() + # 构建返回数据 + page_data = _build_dict_item_page_data(dict_item_list, total, page_size, page_no, pages) + return page_data -def _apply_filters(): +def _apply_dict_item_filters(params: dict): """ 应用查询筛选条件 @@ -489,17 +424,17 @@ def _apply_filters(): query = DictItem.query.filter(DictItem.is_delete == 0) # 字典ID筛选（必填） - dict_id = request.args.get('dictId', 0) + dict_id = params.get("dictId", 0) if dict_id: query = query.filter(DictItem.dict_id == dict_id) # 字典项名称模糊筛选 - name = request.args.get('name') + name = params.get("name") if name: - query = query.filter(DictItem.name.like(like_contains(name), escape='\\')) + query = query.filter(DictItem.name.like(like_contains(name), escape="\\")) # 字典项状态精确筛选 - status = request.args.get('status') + status = params.get("status") if status: query = query.filter(DictItem.status == status) @@ -507,7 +442,7 @@ def _apply_filters(): return query.order_by(DictItem.sort.asc()) -def _build_page_data(dict_item_list, total, page_size, page_no, pages): +def _build_dict_item_page_data(dict_item_list, total, page_size, page_no, pages): """ 构建分页返回数据 @@ -524,19 +459,14 @@ def _build_page_data(dict_item_list, total, page_size, page_no, pages): # 将模型对象列表转换为字典列表 records = models_to_list(dict_item_list) - 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 get_dict_item_detail(dict_item_id): """ 根据ID查询字典项详情（返回字典格式） @@ -552,7 +482,7 @@ def get_dict_item_detail(dict_item_id): """ try: # 复用 get_info 方法查询字典项 - dict_item = get_info(dict_item_id) + dict_item = get_dict_item_info(dict_item_id) if not dict_item: return None # 转换为字典格式 @@ -567,7 +497,8 @@ def get_dict_item_detail(dict_item_id): # 字典项信息查询（根据ID）- 核心查询方法 # ============================================================ -def get_info(dict_item_id: int): + +def get_dict_item_info(dict_item_id: int): """ 根据ID查询字典项信息（返回模型对象） @@ -587,30 +518,14 @@ def get_info(dict_item_id: int): DictItem/None: 字典项模型对象，不存在或已删除时返回 None Example: - dict_item = get_info(dict_item_id) + dict_item = get_dict_item_info(dict_item_id) if dict_item: print(dict_item.name) """ - # 参数有效性验证 - if not dict_item_id or dict_item_id <= 0: - return None - - try: - # 查询未删除的字典项 - dict_item = DictItem.query.filter( - and_(DictItem.id == dict_item_id, DictItem.is_delete == 0) - ).first() - return dict_item - except Exception as e: - logger.error(f"查询字典项信息异常，ID: {dict_item_id}, 错误: {str(e)}") - return None + return DictItemRepository.get_by_id(dict_item_id) -# ============================================================ -# 字典项添加 -# ============================================================ - -def add_dict_item(): +def add_dict_item(data: dict) -> None: """ 添加字典项 @@ -622,42 +537,31 @@ def add_dict_item(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = DictItemForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + # 表单验证 + form = DictItemForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 查询配置（验证配置是否存在） - dict_id = form.dictId.data - dict = Dict.query.filter(Dict.id == dict_id).first() - if not dict: - return R.failed(msg="字典不存在") + # 查询配置（验证配置是否存在） + dict_id = form.dictId.data + dict = Dict.query.filter(Dict.id == dict_id).first() + if not dict: + raise BizError("字典不存在") - # 创建字典项 - dict_item = DictItem(**form.data) - dict_item.dict_code = dict.code - dict_item.create_user = get_username() - dict_item.save() - - return R.ok(msg="添加成功") - - except Exception as e: - logger.error(f"添加字典项异常: {str(e)}") - return R.failed(msg="添加失败，请稍后重试") + # 创建字典项 + dict_item = DictItem(**form.data) + dict_item.dict_code = dict.code + dict_item.create_user = get_username() + dict_item.save() # ============================================================ # 字典项更新 # ============================================================ -def update_dict_item(): + +def update_dict_item(data: dict) -> None: """ 更新字典项 @@ -670,52 +574,42 @@ def update_dict_item(): Returns: R: 操作结果 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - # 表单验证 - form = DictItemForm(data=json_data) - if not form.validate(): - return R.failed(msg=regular.get_err(form)) + # 表单验证 + form = DictItemForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 获取字典项ID - dict_item_id = form.data.get('id') - if not dict_item_id or int(dict_item_id) <= 0: - return R.failed("字典项ID无效") - dict_item_id = int(dict_item_id) + # 获取字典项ID + dict_item_id = form.data.get("id") + if not dict_item_id or int(dict_item_id) <= 0: + raise BizError("字典项ID无效") + dict_item_id = int(dict_item_id) - # 复用 get_info 方法验证字典项是否存在 - dict_item = get_info(dict_item_id) - if not dict_item: - return R.failed("字典项不存在") + # 复用 get_info 方法验证字典项是否存在 + dict_item = get_dict_item_info(dict_item_id) + if not dict_item: + raise BizError("字典项不存在") - # 更新字典项信息（移除ID，避免更新主键） - form_data = convert_camel_to_snake(form.data.copy()) - form_data.pop('id', None) + # 更新字典项信息（移除ID，避免更新主键） + form_data = convert_camel_to_snake(form.data.copy()) + form_data.pop("id", 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 = DictItem.query.filter_by(id=dict_item_id).update(form_data) + result = DictItem.query.filter_by(id=dict_item_id).update(form_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 delete_dict_items(dict_item_ids): + +def delete_dict_items(dict_item_ids: str) -> int: """ 删除字典项（批量删除） @@ -728,4 +622,7 @@ def delete_dict_items(dict_item_ids): Returns: R: 操作结果 """ - return DictItem.batch_delete_r(dict_item_ids) + dict_item_ids_list = parse_id_list(dict_item_ids) + if not dict_item_ids_list: + raise BizError("无效的记录ID") + return DictItem.batch_delete(dict_item_ids_list) diff --git a/src/modules/log/service.py b/src/modules/log/service.py index 6082873..ae9764e 100644 --- a/src/modules/log/service.py +++ b/src/modules/log/service.py @@ -33,6 +33,7 @@ from src.common.constants.dictionary_constants import ( from src.modules.log.models import LoginLog, OperationLog from src.modules.log.login_log_repository import LoginLogRepository from src.modules.log.operation_log_repository import OperationLogRepository +from src.core.database import commit_independently from src.core.exceptions import BizError from src.common.utils.model_utils import models_to_list, model_to_dict from src.common.utils.request import parse_id_list, like_contains @@ -559,7 +560,6 @@ def record_operation_log(title, log_type, source, status, **kwargs): # 使用独立session写入，不受请求事务影响 # 日志是审计数据，无论业务成功或失败都应持久化 - from src.core.database import commit_independently commit_independently(log)