from datetime import datetime, timezone from fastapi.logger import logger from sqlalchemy import and_ from shared.constants import PAGE_SIZE from modules.dictionary.schemas import DictForm from modules.dictionary.models import Dict from core.database import get_db_session from core import R from core.security import get_user_id from utils.request import like_contains # 查询字典分页数据 def get_dict_page(request): db = get_db_session() # 页码 page = int(request.query_params.get("pageNo", 1)) # 每页数 limit = int(request.query_params.get("pageSize", PAGE_SIZE)) # 实例化查询对象 query = db.query(Dict).filter(Dict.is_delete == 0) # 字典名称 name = request.query_params.get('name') if name: query = query.filter(Dict.name.like(like_contains(name), escape='\\')) # 排序 query = query.order_by(Dict.sort.asc()) # 记录总数 count = query.count() # 分页查询 dict_list = query.limit(limit).offset((page - 1) * limit).all() # 若当前页超出末页（数据被并发删除），自动回退到第一页 if page > 1 and not dict_list: page = 1 count = query.count() dict_list = query.limit(limit).offset(0).all() # 实例化结果 result = [] # 遍历数据源 if len(dict_list) > 0: for item in dict_list: # 对象转字典 data = item.to_dict() # 加入列表 result.append(data) # 返回结果 return R.ok(data=result, count=count) # 根据ID查询字典 def get_dict_detail(dict_id): db = get_db_session() # 根据ID查询字典 dict = db.query(Dict).filter(and_(Dict.id == dict_id, Dict.is_delete == 0)).first() # 查询结果判空 if not dict: return None # 对象转字典 data = dict.to_dict() # 返回结果 return data # 添加字典 def add_dict(request, data: DictForm): db = get_db_session() # 唯一性校验 if db.query(Dict).filter(and_(Dict.name == data.name, Dict.is_delete == 0)).first(): return R.failed("字典名称不能重复") # 表单数据赋值给对象 dict = Dict(**data.model_dump(exclude={'id'})) dict.create_user = get_user_id(request) # 插入数据 dict.save() # 返回结果 return R.ok(msg="添加成功") # 更新字典 def update_dict(request, data: DictForm): db = get_db_session() # 记录ID判空 if not data.id or data.id <= 0: return R.failed("记录ID不能为空") # 唯一性校验（排除自身） if db.query(Dict).filter(and_(Dict.name == data.name, Dict.id != data.id, Dict.is_delete == 0)).first(): return R.failed("字典名称不能重复") # 根据ID查询记录 dict = db.query(Dict).filter(and_(Dict.id == data.id, Dict.is_delete == 0)).first() # 查询结果判空 if not dict: return R.failed("记录不存在") # 构造更新数据（排除 id） updData = data.model_dump(exclude={'id'}) updData['update_user'] = get_user_id(request) updData['update_time'] = datetime.now(tz=timezone.utc) result = db.query(Dict).filter_by(id=data.id).update(updData) # 提交数据 if not result: return R.failed("更新失败") # 返回结果 return R.ok(msg="更新成功") # 删除字典 def delete_dict(dict_id): return Dict.batch_delete_r(dict_id) from datetime import datetime, timezone from fastapi.logger import logger from sqlalchemy import and_ from shared.constants import PAGE_SIZE from modules.dictionary.schemas import DictItemForm, DictItemStatusForm from modules.dictionary.models import DictItem from core.database import get_db_session from core import R from core.security import get_user_id from utils.request import like_contains # 查询字典项分页数据 def get_dict_item_page(request): db = get_db_session() # 页码 page = int(request.query_params.get("pageNo", 1)) # 每页数 limit = int(request.query_params.get("pageSize", PAGE_SIZE)) # 实例化查询对象 query = db.query(DictItem).filter(DictItem.is_delete == 0) # 字典ID dict_id = request.query_params.get('dict_id', 0) query = query.filter(DictItem.dict_id == dict_id) # 字典项名称 name = request.query_params.get('name') if name: query = query.filter(DictItem.name.like(like_contains(name), escape='\\')) # 排序 query = query.order_by(DictItem.sort.asc()) # 记录总数 count = query.count() # 分页查询 dict_list = query.limit(limit).offset((page - 1) * limit).all() # 若当前页超出末页（数据被并发删除），自动回退到第一页 if page > 1 and not dict_list: page = 1 count = query.count() dict_list = query.limit(limit).offset(0).all() # 实例化结果 result = [] # 遍历数据源 if len(dict_list) > 0: for item in dict_list: # 对象转字典 data = item.to_dict() # 加入列表 result.append(data) # 返回结果 return R.ok(data=result, count=count) # 根据ID查询字典项详情 def get_dict_item_detail(dict_id): db = get_db_session() # 根据ID查询字典项 dict_data = db.query(DictItem).filter(and_(DictItem.id == dict_id, DictItem.is_delete == 0)).first() # 查询结果判空 if not dict_data: return None # 对象转字典 data = dict_data.to_dict() # 返回结果 return data # 添加字典项 def add_dict_item(request, data: DictItemForm): db = get_db_session() # 唯一性校验 if db.query(DictItem).filter(and_(DictItem.name == data.name, DictItem.is_delete == 0)).first(): return R.failed("字典项名称不能重复") # 表单数据赋值给对象 dict_data = DictItem(**data.model_dump(exclude={'id'})) dict_data.create_user = get_user_id(request) # 插入数据 dict_data.save() # 返回结果 return R.ok(msg="添加成功") # 更新字典项 def update_dict_item(request, data: DictItemForm): db = get_db_session() # 记录ID判空 if not data.id or data.id <= 0: return R.failed("记录ID不能为空") # 唯一性校验（排除自身） if db.query(DictItem).filter(and_(DictItem.name == data.name, DictItem.id != data.id, DictItem.is_delete == 0)).first(): return R.failed("字典项名称不能重复") # 根据ID查询记录 dict_data = db.query(DictItem).filter(and_(DictItem.id == data.id, DictItem.is_delete == 0)).first() # 查询结果判空 if not dict_data: return R.failed("记录不存在") # 构造更新数据（排除 id） updData = data.model_dump(exclude={'id'}) updData['update_user'] = get_user_id(request) updData['update_time'] = datetime.now(tz=timezone.utc) result = db.query(DictItem).filter_by(id=data.id).update(updData) # 提交数据 if not result: return R.failed("更新失败") # 返回结果 return R.ok(msg="更新成功") # 删除字典项 def delete_dict_item(dict_id): return DictItem.batch_delete_r(dict_id) # 设置状态 def update_dict_item_status(request, data: DictItemStatusForm): db = get_db_session() # 根据ID查询记录 dict_data = db.query(DictItem).filter(and_(DictItem.id == data.id, DictItem.is_delete == 0)).first() # 查询结果判空 if not dict_data: return R.failed("记录不存在") # 更新记录 result = db.query(DictItem).filter_by(id=data.id).update({ "status": data.status, "update_user": get_user_id(request), "update_time": datetime.now(tz=timezone.utc) }) # 提交数据 if not result: return R.failed("设置失败") # 返回结果 return R.ok(msg="设置成功")