djangoadmin/DjangoAdmin_Flask_EleVue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: P4c 批次1 服务层改造（notice/param/link/level）

Browse Source 
【服务层新范式批量铺开】4 个 uniform 模块：
- service：get_x_page(params)->dict、add/update/status(data)->None、
  delete(ids)->int、get_info 委托 Repository（新建 repository.py）
- 业务失败 raise BizError；router 取参包 R
- level 的 Excel import/export 暂保留 R 返回（后续统一改造）

【工具】codemod 函数级转换（try/except 退格移除、R.failed→BizError、
R.ok→return data）+ black 格式化 + 未用导入清理

【教训】清理脚本 startswith 前缀误删含多名字的 import 行（models_to_list,
model_to_dict 同行）导致 F821，已按 F821 精确补回

【验收】应用启动；4 服务分页查询运行时正常；12 passed；flake8 零告警

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
2026-08-07 15:15:22 +08:00
co-authored by Claude
parent 63afa4e629
commit 3f55f62f0d
17 changed files with 570 additions and 656 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/system/level/service.py
+95 -146
View File
@@ -71,24 +71,25 @@ import time
import logging
from datetime import datetime, timezone
from flask import request
from sqlalchemy import and_
from openpyxl.reader.excel import load_workbook
from openpyxl.workbook import Workbook
from sqlalchemy import and_
import utils
from src.common.constants.common import PAGE_SIZE
from src.modules.system.level.schema import LevelForm, LevelStatusForm
from src.modules.system.level.models import Level
from src.modules.system.level.repository import LevelRepository
from src.core import response as R
from src.core.database import db
from src.core.exceptions import BizError
from src.modules.upload import service as upload
from config import FLASK_TEMP_PATH, FLASK_UPLOAD_DIR
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.common.utils.request import parse_id_list, like_contains
from src.common.utils.rich_text import get_file_url
from src.common.utils.file import mkdir
from src.core.security import get_username
@@ -101,7 +102,8 @@ logger = logging.getLogger(__name__)
# 职级分页查询
# ============================================================
def get_level_page():
def get_level_page(params: dict) -> dict:
"""
查询职级分页数据
@@ -122,31 +124,26 @@ def get_level_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
# 分页查询
level_list = query.limit(page_size).offset((page_no - 1) * page_size).all()
# 构建返回数据
page_data = _build_page_data(level_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="查询失败，请稍后重试")
# 分页查询
level_list = query.limit(page_size).offset((page_no - 1) * page_size).all()
# 构建返回数据
page_data = _build_page_data(level_list, total, page_size, page_no, pages)
return page_data
def _apply_filters():
def _apply_filters(params: dict):
"""
应用查询筛选条件
@@ -163,12 +160,12 @@ def _apply_filters():
query = Level.query.filter(Level.is_delete == 0)
# 职级名称模糊筛选
name = request.args.get('name')
name = params.get("name")
if name:
query = query.filter(Level.name.like(like_contains(name), escape='\\'))
query = query.filter(Level.name.like(like_contains(name), escape="\\"))
# 职级状态精确筛选
status = request.args.get('status')
status = params.get("status")
if status:
query = query.filter(Level.status == status)
@@ -193,19 +190,14 @@ def _build_page_data(level_list, total, page_size, page_no, pages):
# 将模型对象列表转换为字典列表
records = models_to_list(level_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_level_detail(level_id):
"""
根据ID查询职级详情（返回字典格式）
@@ -236,6 +228,7 @@ def get_level_detail(level_id):
# 职级信息查询（根据ID）- 核心查询方法
# ============================================================
def get_info(level_id: int):
"""
根据ID查询职级信息（返回模型对象）
@@ -261,26 +254,10 @@ def get_info(level_id: int):
if level:
print(level.name)
"""
# 参数有效性验证
if not level_id or level_id <= 0:
return None
try:
# 查询未删除的职级
level = Level.query.filter(
and_(Level.id == level_id, Level.is_delete == 0)
).first()
return level
except Exception as e:
logger.error(f"查询职级信息异常，ID: {level_id}, 错误: {str(e)}")
return None
return LevelRepository.get_by_id(level_id)
# ============================================================
# 职级添加
# ============================================================
def add_level():
def add_level(data: dict) -> None:
"""
添加职级
@@ -292,35 +269,24 @@ def add_level():
Returns:
R: 操作结果
"""
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证
form = LevelForm(data=json_data)
if not form.validate():
err_msg = regular.get_err(form)
return R.failed(msg=err_msg)
# 表单验证
form = LevelForm(data=data)
if not form.validate():
raise BizError(regular.get_err(form))
# 创建职级
level = Level(**form.data)
level.create_user = get_username()
level.save()
return R.ok(msg="添加成功")
except Exception as e:
logger.error(f"添加职级异常: {str(e)}")
return R.failed(msg="添加失败，请稍后重试")
# 创建职级
level = Level(**form.data)
level.create_user = get_username()
level.save()
# ============================================================
# 职级更新
# ============================================================
def update_level():
def update_level(data: dict) -> None:
"""
更新职级
@@ -333,52 +299,42 @@ def update_level():
Returns:
R: 操作结果
"""
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证
form = LevelForm(data=json_data)
if not form.validate():
return R.failed(msg=regular.get_err(form))
# 表单验证
form = LevelForm(data=data)
if not form.validate():
raise BizError(regular.get_err(form))
# 获取职级ID
level_id = form.data.get('id')
if not level_id or int(level_id) <= 0:
return R.failed("职级ID无效")
level_id = int(level_id)
# 获取职级ID
level_id = form.data.get("id")
if not level_id or int(level_id) <= 0:
raise BizError("职级ID无效")
level_id = int(level_id)
# 复用 get_info 方法验证职级是否存在
level = get_info(level_id)
if not level:
return R.failed("职级不存在")
# 复用 get_info 方法验证职级是否存在
level = get_info(level_id)
if not level:
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 = Level.query.filter_by(id=level_id).update(form_data)
result = Level.query.filter_by(id=level_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_levels(level_ids):
def delete_levels(level_ids: str) -> int:
"""
删除职级（批量删除）
@@ -391,14 +347,18 @@ def delete_levels(level_ids):
Returns:
R: 操作结果
"""
return Level.batch_delete_r(level_ids)
level_ids_list = parse_id_list(level_ids)
if not level_ids_list:
raise BizError("无效的记录ID")
return Level.batch_delete(level_ids_list)
# ============================================================
# 职级状态设置
# ============================================================
def update_level_status():
def update_level_status(data: dict) -> None:
"""
设置职级状态
@@ -407,44 +367,33 @@ def update_level_status():
Returns:
R: 操作结果
"""
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证
form = LevelStatusForm(data=json_data)
if not form.validate():
err_msg = regular.get_err(form)
return R.failed(msg=err_msg)
# 表单验证
form = LevelStatusForm(data=data)
if not form.validate():
raise BizError(regular.get_err(form))
level_id = int(form.data['id'])
status = int(form.data['status'])
level_id = int(form.data["id"])
status = int(form.data["status"])
# 复用 get_info 方法验证职级是否存在
level = get_info(level_id)
if not level:
return R.failed("职级不存在")
# 复用 get_info 方法验证职级是否存在
level = get_info(level_id)
if not level:
raise BizError("职级不存在")
# 更新状态
update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()}
result = Level.query.filter_by(id=level_id).update(update_data)
# 更新状态
update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()}
result = Level.query.filter_by(id=level_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("设置失败")
# ============================================================
# Excel导入
# ============================================================
def import_levels():
"""
导入Excel文件批量导入职级
@@ -462,7 +411,7 @@ def import_levels():
try:
# 上传并读取Excel文件
file = upload.upload_file(".xls|.xlsx")
path = file['filePath']
path = file["filePath"]
wb = load_workbook(path)
ws = wb.worksheets[0]
@@ -502,8 +451,8 @@ def import_levels():
finally:
# 清理导入产生的临时Excel文件
try:
if 'path' in dir() and path:
abs_path = os.path.join(FLASK_UPLOAD_DIR, path.lstrip('/'))
if "path" in dir() and path:
abs_path = os.path.join(FLASK_UPLOAD_DIR, path.lstrip("/"))
if os.path.isfile(abs_path):
os.remove(abs_path)
except Exception:
@@ -514,6 +463,7 @@ def import_levels():
# Excel导出
# ============================================================
def export_levels():
"""
导出职级数据到Excel文件
@@ -533,17 +483,17 @@ def export_levels():
wb = Workbook()
ws = wb.worksheets[0]
# 写入表头
ws.append(['职级ID', '职级名称', '职级排序', '职级状态', '创建时间', '更新时间'])
ws.append(["职级ID", "职级名称", "职级排序", "职级状态", "创建时间", "更新时间"])
# 写入数据
for data in level_list:
ws.append(list(data.values()))
# 保存文件
save_path = FLASK_TEMP_PATH + "/" + time.strftime('%Y%m%d')
save_path = FLASK_TEMP_PATH + "/" + time.strftime("%Y%m%d")
mkdir(save_path)
file_name = time.strftime('%Y%m%d%H%M%S') + '%05d' % random.randint(0, 100) + ".xlsx"
file_name = time.strftime("%Y%m%d%H%M%S") + "%05d" % random.randint(0, 100) + ".xlsx"
file_path = save_path + "/" + file_name
wb.save(file_path)
@@ -562,6 +512,7 @@ def export_levels():
# 职级列表获取（无分页）
# ============================================================
def get_level_list():
"""
获取所有启用的职级列表（无分页）
@@ -574,9 +525,7 @@ def get_level_list():
"""
try:
# 查询所有启用的职级
level_list = Level.query.filter(
and_(Level.is_delete == 0, Level.status == 1)
).order_by(Level.sort.asc()).all()
level_list = Level.query.filter(and_(Level.is_delete == 0, Level.status == 1)).order_by(Level.sort.asc()).all()
# 转换为字典列表
result = []