Service 层

Service 层是业务逻辑层，负责实现具体的业务规则。简单 CRUD 模块继承 BaseService 基类，只需声明差异点即可获得完整的增删改查能力。

文件位置

src/modules/system/position/service.py

完整代码

from typing import Optional

from core.base_service import BaseService
from modules.system.position.models import Position
from modules.system.position.repository import position_repo
from modules.system.user.models import User
from modules.system.user.repository import user_repo
from utils.request import parse_id_list


class PositionService(BaseService[Position]):
    """岗位业务服务类，继承基础服务获得通用 CRUD 能力"""
    # 数据访问与模型
    repo = position_repo
    model = Position
    # 分页差异点
    page_like_fields = ('name',)
    page_eq_fields = ('status',)
    page_order_by = (('sort', 'asc'),)
    # 唯一性校验
    unique_fields = {'name': '岗位名称不能重复'}

    def _before_delete(self, ids) -> Optional[str]:
        """删除前校验：存在用户引用该岗位时禁止删除"""
        id_list = parse_id_list(str(ids))
        if id_list and user_repo.filter(User.position_id.in_(id_list), User.is_delete == 0).first():
            return "存在用户引用该岗位，请先调整用户岗位"
        return None

    def get_position_list(self, request):
        """获取岗位数据列表（无分页，用于下拉选择）"""
        query = self._apply_page_filters(self.repo.filter_by(), request)
        query = query.order_by(self.model.sort.asc())
        return [v.to_dict() for v in query.all()]


# 模块级单例
position_service = PositionService()

代码解析

必须声明的差异点

repo = position_repo      # 关联的 Repository 实例
model = Position           # 关联的模型类

分页查询配置

page_like_fields = ('name',)          # 模糊查询字段（LIKE）
page_eq_fields = ('status',)          # 精确查询字段（=）
page_order_by = (('sort', 'asc'),)    # 排序规则

当调用 get_page(request) 时，基类自动：

1. 从 request.query_params 读取 pageNo、pageSize
2. 对 page_like_fields 中的字段执行 LIKE %keyword% 查询
3. 对 page_eq_fields 中的字段执行精确匹配
4. 按 page_order_by 排序
5. 返回分页数据

唯一性校验

unique_fields = {'name': '岗位名称不能重复'}

字典的 key 是模型字段名，value 是校验失败时的错误提示。在 add() 和 update() 时，基类自动查询数据库判断是否重复。

生命周期钩子

def _before_delete(self, ids) -> Optional[str]:

BaseService 提供以下钩子方法，子类可按需覆盖：

钩子	触发时机	返回值
_before_delete(ids)	删除操作前	返回错误字符串则中止删除，返回 None 则继续
_before_create(data)	创建操作前	可修改 data
_before_update(data, existing)	更新操作前	可修改 data
_serialize(item)	列表序列化时	自定义返回字段
_serialize_detail(item)	详情序列化时	自定义返回字段

自定义业务方法

def get_position_list(self, request):
    """获取岗位数据列表（无分页）"""
    query = self._apply_page_filters(self.repo.filter_by(), request)
    query = query.order_by(self.model.sort.asc())
    return [v.to_dict() for v in query.all()]

_apply_page_filters 是基类方法，复用分页的筛选逻辑（模糊 + 精确），但不执行分页，返回全部匹配结果。

基类提供的通用方法

方法	说明
get_page(request)	分页查询
get_detail(id)	查询详情
add(request, data)	新增记录
update(request, data)	更新记录
delete(id)	单条删除
batch_delete(request)	批量删除（需 await）
update_status(request, data)	更新状态

开发要点

1. 必须声明 repo 和 model：基类所有方法依赖这两个属性
2. 分页配置按需声明：page_like_fields / page_eq_fields / page_order_by
3. 唯一性校验通过 unique_fields 配置：基类自动处理，无需手写查询
4. 删除前校验用 _before_delete 钩子：返回错误字符串中止操作
5. 必须创建模块级单例：xxx_service = XxxService()
6. 删除操作使用 BaseService.batch_delete_with_r(repo, ids)：统一响应文案

总结

Service 层继承 BaseService 获得通用 CRUD 能力，只需声明差异点（repo/model/page_like_fields/unique_fields/serialize_maps）并覆写钩子方法。删除操作使用 BaseService.batch_delete_with_r() 统一响应文案，必须创建模块级单例。