分层架构设计

本章详细描述项目的三层架构设计，包括各层的职责边界、依赖方向、通信方式，并以岗位（Position）模块为实例贯穿说明。

三层架构概览

架构分层原则

采用经典的三层架构，核心原则是：依赖单向、职责单一、接口隔离。上层只依赖下层接口，下层不感知上层存在。

┌─────────────────────────────────────────────────────────────┐
│ HTTP 层 (Router)                                            │
│ modules/system/position/router.py                           │
│ 职责：路由定义 · 权限装饰器 · 参数接收 · 响应返回           │
│ 原则：不含业务逻辑，仅做请求转发                             │
└──────────────────────────┬──────────────────────────────────┘
                           │ 调用
                           ▼
┌─────────────────────────────────────────────────────────────┐
│ 业务层 (Service)                                            │
│ modules/system/position/service.py                          │
│ 职责：业务逻辑 · 校验规则 · 字段组装 · 序列化加工           │
│ 原则：不直接操作数据库，通过 Repository 访问数据             │
└──────────────────────────┬──────────────────────────────────┘
                           │ 调用
                           ▼
┌─────────────────────────────────────────────────────────────┐
│ 数据层 (Repository)                                         │
│ modules/system/position/repository.py                       │
│ 职责：数据库 CRUD · 查询条件组装 · 软删除过滤               │
│ 原则：只做数据访问原语，不做业务校验和响应封装               │
└──────────────────────────┬──────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│ 数据库 (MySQL / PostgreSQL / etc.)                          │
└─────────────────────────────────────────────────────────────┘

各层详细职责

第一层：HTTP 层（Router）

目录位置：src/modules/{group}/{name}/router.py

核心职责：

1. 路由定义：声明 HTTP 方法、URL 路径、请求参数类型
2. 权限控制：通过 @check_permission 装饰器声明所需权限
3. 参数接收：通过 parse_request_body 函数解析请求体
4. 响应返回：直接返回 Service 层的 R 对象

设计原则：

• 不包含任何业务逻辑
• 不直接访问数据库
• 仅做请求转发和响应返回
• 权限字符串遵循 sys:{module}:{action} 格式

代码示例（岗位模块）：

# src/modules/system/position/router.py
# +======================================================================
# | 模块: 岗位路由层
# | 说明: 提供岗位信息的增删改查、状态管理、列表获取等 API 接口
# +======================================================================

from flask import Blueprint, request

from src.core.permission import check_permission
from src.core import response as R
from src.common.utils.request import parse_request_body, parse_batch_ids

# ============================================================
# 创建 Blueprint 实例
# ============================================================
position_blue = Blueprint('position', __name__, url_prefix='/position')


# ============================================================
# 查询分页数据
# ============================================================
@position_blue.route('/page', methods=['GET'])
@check_permission('sys:position:page')
def page():
    """
    获取岗位分页列表
    :return: 分页数据响应
    """
    from src.modules.system.position.service import get_position_page
    return get_position_page()


# ============================================================
# 查询岗位详情
# ============================================================
@position_blue.route('/detail/<int:id>', methods=['GET'])
@check_permission('sys:position:detail')
def detail(id):
    """
    根据岗位ID查询详细信息
    :param id: 岗位ID（路径参数）
    :return: 岗位详情数据
    """
    from src.modules.system.position.service import get_position_detail
    return R.ok(data=get_position_detail(id))


# ============================================================
# 获取岗位列表
# ============================================================
@position_blue.route('/list', methods=['GET'])
@check_permission('sys:position:list')
def get_position_list():
    """
    获取所有岗位列表（无分页，用于下拉选择）
    :return: 岗位列表数据
    """
    from src.modules.system.position.service import get_position_list
    return R.ok(data=get_position_list())


# ============================================================
# 添加岗位
# ============================================================
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
def add():
    """
    新增岗位
    :return: 添加结果响应
    """
    from src.modules.system.position.service import add_position
    from src.modules.system.position.schema import PositionForm
    data = parse_request_body(request, PositionForm)
    return add_position(data)


# ============================================================
# 更新岗位
# ============================================================
@position_blue.route('/update', methods=['PUT'])
@check_permission('sys:position:update')
def update():
    """
    更新岗位信息
    :return: 更新结果响应
    """
    from src.modules.system.position.service import update_position
    from src.modules.system.position.schema import PositionForm
    data = parse_request_body(request, PositionForm)
    return update_position(data)


# ============================================================
# 删除岗位
# ============================================================
@position_blue.route('/delete/<int:id>', methods=['DELETE'])
@check_permission('sys:position:delete')
def delete(id):
    """
    根据ID删除单个岗位
    :param id: 岗位ID（路径参数）
    :return: 删除结果响应
    """
    from src.modules.system.position.service import delete_position
    return delete_position(id)


# ============================================================
# 设置岗位状态
# ============================================================
@position_blue.route('/status', methods=['PUT'])
@check_permission('sys:position:status')
def status():
    """
    修改岗位状态（启用/禁用）
    :return: 状态更新结果响应
    """
    from src.modules.system.position.service import update_position_status
    from src.modules.system.position.schema import PositionStatusForm
    data = parse_request_body(request, PositionStatusForm)
    return update_position_status(data)


# ============================================================
# 批量删除岗位
# ============================================================
@position_blue.route('/batchDelete', methods=['DELETE'])
@check_permission('sys:position:batchDelete')
def batch_delete():
    """
    批量删除岗位（请求体需传入 ID 数组）

    Request Body 示例:
        [1, 2, 3]  # 岗位ID数组

    :return: 批量删除结果响应
    """
    from src.modules.system.position.service import delete_positions
    id_list = parse_batch_ids(request)
    return delete_positions(id_list)

装饰器顺序

装饰器按从上到下执行：先 @position_blue.route()（路由），再 @check_permission（鉴权）。

第二层：业务层（Service）

目录位置：src/modules/{group}/{name}/service.py

核心职责：

1. 业务逻辑编排：组合多个数据操作完成业务流程
2. 唯一性校验：通过 Schema 的 validate_ 方法 + Repository.exists_by() 实现
3. 表单校验：_validate_form 自动校验，_clean_form_data 过滤无效字段
4. 序列化加工：_detail_dict / _enrich_page_records 加工响应数据
5. 文件处理：文件字段迁移、URL 补全
6. 前置/后置钩子：before_add / before_update / before_delete / after_add / after_update / after_delete 扩展点

设计原则：

• 通过 Repository 访问数据库，不直接使用 db.query()
• 返回 R 对象，不做 jsonify 封装
• 通过声明差异点定制通用 CRUD，减少重复代码

代码示例（岗位模块）：

# src/modules/system/position/service.py
# +======================================================================
# | 模块: 岗位业务逻辑层
# | 说明: 岗位的增删改查、唯一性校验、状态管理等业务处理
# +======================================================================

from typing import Optional

from src.common.base_service import BaseService
from src.modules.system.position.models import Position
from src.modules.system.position.repository import PositionRepository
from src.modules.system.user.models import User
from src.modules.system.user.repository import UserRepository


class PositionService(BaseService):
    """
    岗位服务类

    继承 BaseService 获得标准 CRUD 能力，并扩展删除前用户引用校验。
    """
    repository = PositionRepository
    form = PositionForm
    status_form = PositionStatusForm
    model_name = "岗位"

    @classmethod
    def before_delete(cls, ids):
        """删除前校验：存在用户引用该岗位时禁止删除"""
        for pid in ids:
            if UserRepository.count_by({'position_id': pid}) > 0:
                raise BizError(f"岗位ID {pid} 已被用户引用，无法删除")


# ============================================================
# 模块级代理函数：保持 router 调用兼容
# ============================================================
get_position_page = PositionService.get_page
get_position_detail = PositionService.get_detail
get_info = PositionService.get_info
add_position = PositionService.add
update_position = PositionService.update
delete_positions = PositionService.remove
update_position_status = PositionService.update_status
get_position_list = PositionService.get_list

Service 层的两种模式

简单 CRUD 模块继承 BaseService，只需声明 repository / form / model_name 并覆写钩子方法。复杂模块（如用户）在继承基础上覆写 add / update 等方法。

第三层：数据层（Repository）

目录位置：src/modules/{group}/{name}/repository.py

核心职责：

1. CRUD 操作：create / update / batch_delete / get_by_id 等
2. 查询构建：filter / filter_by / get_one / get_all / count
3. 分页查询：paginate 返回 (items, total) 元组
4. 软删除过滤：等值方法自动追加 is_delete=0
5. 数据清洗：_clean_form_data 过滤无效字段，convert_camel_to_snake 转命名

设计原则：

• 只做数据访问原语，不做 R 封装
• 等值方法自动软删过滤，filter() 为原始出口
• 删除的 R 响应由 Service 层统一封装

代码示例（岗位模块）：

# src/modules/system/position/repository.py
from src.common.base_repository import BaseRepository
from src.common.constants.common import PAGE_SIZE
from src.modules.system.position.models import Position


class PositionRepository(BaseRepository):
    """岗位数据仓储"""
    model = Position

    @classmethod
    def page(cls, params: dict):
        """岗位分页查询"""
        return cls.find_page(
            page=int(params.get('pageNo', 1)),
            page_size=int(params.get('pageSize', PAGE_SIZE)),
            filters={'status': params.get('status')},
            like_filters={'name': params.get('name')},
            order_by='sort',
        )

    @classmethod
    def list_enabled(cls, params=None):
        """查询启用的岗位列表"""
        params = params or {}
        return cls.find_list(
            filters={'status': 1},
            like_filters={'name': params.get('name')},
            order_by='sort',
        )

Repository 基类提供的方法

BaseRepository 提供以下通用方法，所有方法均为 classmethod：

• 查询：get_by_id / get_by_ids / first_by / exists_by / count_by / exists
• 分页：find_page 返回 (items, total, pages)
• 列表：find_list 返回实例列表
• 写操作：create / update_by_id / update_status / delete_by_id / batch_delete / batch_restore

依赖方向

HTTP 层 ──依赖──→ Service 层 ──依赖──→ Repository 层 ──依赖──→ Model
 │                │                    │
 │                │                    │
 ▼                ▼                    ▼
R (响应封装)      R (响应封装)         BaseRepository (基类)
WTForms           BaseRepository       db.session
check_permission  BaseService          SQLAlchemy ORM

依赖规则

• HTTP 层只依赖 Service 层，不直接调用 Repository
• Service 层通过 Repository 访问数据库，不直接使用 db.query()
• Repository 层只依赖 Model 和数据库会话
• 下层不感知上层存在，不调用上层方法

模型层（Model）

目录位置：src/modules/{group}/{name}/models.py

模型层定义数据库表结构，继承 BaseModel（通用字段）和 BaseDB（持久化方法）。

# src/modules/system/position/models.py
from sqlalchemy import Column, String, Integer, text

from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX


# ============================================================
# 岗位模型
# ============================================================
class Position(BaseModel, BaseDB):
    """岗位模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "position"
    __table_comment__ = "岗位表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 岗位名称
    name = Column(String(255), nullable=False, index=True, comment="岗位名称")
    # 岗位状态：1-在用 2-停用
    status = Column(Integer, default=0, server_default=text('0'), index=True, comment="岗位状态：1-在用 2-停用")
    # 岗位排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="岗位排序")

    # ============================================================
    # 内置方法
    # ============================================================
    def __repr__(self):
        """返回岗位名称作为字符串表示"""
        return '岗位：{}'.format(self.name)

Schema 层（校验）

目录位置：src/modules/{group}/{name}/schema.py

Schema 层定义请求参数的校验规则，使用 WTForms 的表单类 + validators。

# src/modules/system/position/schema.py
# +======================================================================
# | 模块: 岗位表单验证
# | 说明: 岗位创建/编辑/状态更新的请求数据校验
# +======================================================================

from wtforms import StringField, IntegerField
from wtforms.validators import DataRequired, Length, NumberRange

from src.common.forms.base_form import BaseForm


# ============================================================
# 岗位表单类
# ============================================================
class PositionForm(BaseForm):
    """岗位创建/编辑表单"""
    name = StringField('岗位名称', validators=[
        DataRequired(message='岗位名称不能为空'),
        Length(min=1, max=150, message='岗位名称长度必须在1-150个字符之间')
    ])
    status = IntegerField('岗位状态', validators=[
        DataRequired(message='岗位状态不能为空'),
        NumberRange(min=1, max=2, message='岗位状态必须是1或2')
    ])
    sort = IntegerField('岗位排序', validators=[
        DataRequired(message='岗位排序不能为空'),
        NumberRange(min=0, max=99999, message='岗位排序必须在0-99999之间')
    ])


# ============================================================
# 设置状态表单类
# ============================================================
class PositionStatusForm(BaseForm):
    """岗位状态更新表单"""
    id = IntegerField('岗位ID', validators=[
        DataRequired(message='岗位ID不能为空'),
        NumberRange(min=1, message='岗位ID必须大于0')
    ])
    status = IntegerField('岗位状态', validators=[
        DataRequired(message='岗位状态不能为空'),
        NumberRange(min=1, max=2, message='岗位状态必须是1或2')
    ])

完整调用链路示例

以添加岗位为例，展示三层之间的调用关系：

POST /api/v1/position/add
│
├─ 1. Router (position.py)
│   @check_permission('sys:position:add')
│   def add():
│       data = parse_request_body(request, PositionForm)
│       return add_position(data)
│
├─ 2. Service (service.py)
│   @classmethod
│   def add(cls, data):
│       form = cls._validate_form(data)     # 表单校验
│       extra = cls.before_add(form) or {}  # 前置钩子
│       create_data = cls._clean_form_data(form)
│       create_data.update(extra)
│       obj = cls.repository.create(**create_data)
│       cls.after_add(obj)                  # 后置钩子
│
├─ 3. Repository (base_repository.py)
│   @classmethod
│   def create(cls, **kwargs):
│       obj = cls.model(**kwargs)
│       obj.save()                          # add + flush
│       return obj
│
├─ 4. Model (base_db.py)
│   def save(self):
│       db.session.add(self)
│       db.session.flush()                  # SQL 发送到数据库
│
└─ 5. Database
        INSERT INTO flask_position (name, status, sort, create_user, ...)
        VALUES ('高级工程师', 1, 10, '管理员', ...)

各层文件职责一览

文件	层次	职责	是否包含业务逻辑
position/router.py	HTTP	路由、权限、参数接收	否
position/schema.py	HTTP	请求参数校验规则	否
position/service.py	业务	业务逻辑、校验、序列化	是
position/repository.py	数据	数据库 CRUD、查询构建	否
position/models.py	数据	表结构定义、字段映射	否

总结

三层架构通过依赖单向和职责单一实现了良好的关注点分离。HTTP 层只关心请求接收和响应返回，Service 层只关心业务逻辑编排，Repository 层只关心数据访问。这种分层使得每一层都可以独立测试、独立替换，系统在模块数量增长时仍能保持清晰的代码结构。以岗位模块为例，完整实现一个 CRUD 模块只需 5 个文件，其中 Repository 和 Model 几乎零代码，开发者只需关注 Service 层的业务差异点。