Service层规范

概述

Service 层是业务逻辑层，各模块 Service 继承 BaseService[Model] 获得通用 CRUD 能力。子类只需声明差异点（repo、model、过滤字段、排序、唯一性等），即可获得完整的增删改查接口。

设计模式

BaseService 采用模板方法模式：基类定义流程骨架，子类通过声明差异点和覆盖钩子定制行为。

PositionService 示例

以岗位模块为例，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):
        """获取岗位数据列表（不分页，供下拉选择使用）"""
        # 与 /page 一致的筛选条件：name 模糊、status 精确；保留软删过滤与排序
        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	BaseRepository	数据仓库实例	position_repo
model	Model	ORM 模型类	Position
page_like_fields	tuple	分页模糊查询字段	('name',)
page_eq_fields	tuple	分页等值查询字段	('status',)
page_order_by	tuple	分页排序规则	(('sort', 'asc'),)
unique_fields	dict	唯一性校验字段	{'name': '名称重复'}
serialize_maps	dict	枚举显示名映射	{'status': {1: '在用', 2: '停用'}}
file_fields	tuple	文件字段	('cover', 'image')
rich_text_fields	tuple	富文本字段	('content',)

单例模式

每个模块在文件末尾创建 Service 实例（如 position_service），供 Endpoint 层使用。

可覆盖钩子

查询钩子

def _apply_page_filters(self, query, request):
    """组装分页过滤条件"""
    # 调用基类默认实现（处理 page_like_fields / page_eq_fields）
    query = super()._apply_page_filters(query, request)
    # 追加固定过滤
    query = query.filter(self.model.status == 1)
    return query

def _serialize(self, item) -> Dict[str, Any]:
    """列表行序列化"""
    data = item.to_dict()
    # 自定义序列化逻辑
    return data

def _serialize_detail(self, obj) -> Dict[str, Any]:
    """详情序列化"""
    data = self._serialize(obj)
    # 详情特有字段
    return data

写操作钩子

def _before_add(self, request, data) -> None:
    """新增前扩展（如密码哈希）"""

def _before_update(self, request, data) -> None:
    """编辑前扩展"""

def _before_delete(self, ids) -> Optional[str]:
    """删除前校验：返回 None 放行，返回字符串拦截"""

def _build_create_fields(self, request, data) -> Dict[str, Any]:
    """组装新增字段"""
    fields = super()._build_create_fields(request, data)
    # 追加自定义字段
    return fields

def _build_update_fields(self, request, data) -> Dict[str, Any]:
    """组装更新字段"""
    fields = super()._build_update_fields(request, data)
    # 追加自定义字段
    return fields

batch_delete_with_r 静态方法

@staticmethod
def batch_delete_with_r(repo: BaseRepository, ids) -> R:
    """按逗号分隔ID软删除并返回标准R响应"""
    id_list = parse_id_list(str(ids))
    if not id_list:
        return R.failed("记录ID不存在")

    count = repo.batch_delete(ids)
    if count != len(id_list):
        return R.failed("记录不存在")

    return R.ok(msg="本次共删除{0}条数据".format(count))

复用规则

自定义 Service 的删除方法应复用 BaseService.batch_delete_with_r(repo, ids)，保证删除响应文案全仓一致。禁止直接调用 repo.batch_delete 后自行封装 R。

同步/异步约定

Service 方法	同步/异步	原因
get_page / get_detail / add / update / delete	同步 def	纯 DB 操作
batch_delete	异步 async def	需 await parse_batch_ids(request)
含 Redis 缓存的方法	异步 async def	Redis IO 需 await

禁止混用

• 同步 def Service 中禁止直接调用异步方法
• 异步 async def Endpoint 中禁止直接调用同步 DB 查询（用 asyncio.to_thread 包裹）

serialize_maps 枚举映射

class ArticleService(BaseService[Article]):
    # 字典编码方式（推荐）
    serialize_maps = {
        'source': 'article_source',  # 从数据字典取 {value: label}
    }

    # 字面字典方式（历史兼容）
    serialize_maps = {
        'status': {1: '在用', 2: '停用'},
    }

序列化结果

serialize_maps 中的字段会在序列化时自动补一个 {field}Text 字段。例如 source=1 会生成 sourceText="原创"。

总结

Service 层通过继承 BaseService[Model] 获得通用 CRUD 能力。子类声明差异点（repo/model/page_like_fields/page_eq_fields/page_order_by/unique_fields/serialize_maps），覆盖钩子（_apply_page_filters/_before_delete/_serialize 等）定制业务行为。删除操作统一复用 batch_delete_with_r 静态方法。模块级单例模式供 Endpoint 调用。