核心基类体系

本章详细描述项目的核心基类体系，包括 BaseModel、BaseDB、BaseForm、BaseRepository、BaseService 五大基类的设计理念、继承关系和使用方式。以岗位（Position）模块为实例贯穿说明。

基类继承关系

基类体系总览

五大基类各司其职，覆盖从数据模型到请求校验再到业务逻辑的完整链路。子类只需声明差异点，即可获得标准化的增删改查能力。

┌──────────────────────────────────────────────────────────────────┐
│ SQLAlchemy Base (declarative_base)                               │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────────┐
│ BaseModel (src/common/base_model.py)                             │
│ ─────────────────────────────────────────────────────────────── │
│ 通用字段: id / create_user / create_time / update_user /         │
│ update_time / is_delete                                          │
│ 序列化: to_dict() → 驼峰字段名 + UTC 转本地 + 排除敏感字段       │
└──────────────────────────┬───────────────────────────────────────┘
                           │ 多继承
                           ▼
┌──────────────────────────────────────────────────────────────────┐
│ BaseDB (src/common/base_db.py)                                   │
│ ─────────────────────────────────────────────────────────────── │
│ 持久化: save() → db.session.add + db.session.flush（不 commit）  │
│ 事务: 由 AOP 事务中间件统一管理                                  │
└──────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                Position(BaseModel, BaseDB)
                ┌──────────────────────────┐
                │ __tablename__            │
                │ name / status / sort     │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseForm (src/common/forms/base_form.py)                         │
│ ─────────────────────────────────────────────────────────────── │
│ WTForms 表单基类                                                 │
│ 可选 id 字段: Optional，创建时可省略                             │
│ 空字符串容错: '' / 'null' / 'undefined' → None                   │
└──────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                PositionForm(BaseForm)
                ┌──────────────────────────┐
                │ name / status / sort     │
                │ 继承可选 id              │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseRepository (src/common/base_repository.py)                   │
│ ─────────────────────────────────────────────────────────────── │
│ 查询: get_by_id / get_by_ids / get_all / first_by / count_by    │
│ 分页: find_page / find_list                                      │
│ 写操作: create / update / batch_delete                           │
│ 条件: exists / exists_by (等值+软删)                             │
│ 软删除: 等值方法自动追加 is_delete=0                             │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                PositionRepository(BaseRepository)
                ┌──────────────────────────┐
                │ model = Position         │
                │ _repository = 单例       │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseService (src/common/base_service.py)                         │
│ ─────────────────────────────────────────────────────────────── │
│ 查询: get_page / get_list / get_detail / get_info                │
│ 写操作: add / update / remove / update_status                    │
│ 差异点: repository / form / status_form / model_name             │
│ 钩子: before_add / before_update / before_delete                 │
│ after_add / after_update / after_delete                          │
│ _enrich_page_records / _detail_dict                              │
│ _validate_form / _clean_form_data                                │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                PositionService(BaseService)
                ┌──────────────────────────────────┐
                │ repository = PositionRepository  │
                │ form = PositionForm              │
                │ status_form = PositionStatusForm │
                │ model_name = "岗位"              │
                │ before_delete() → 引用检查       │
                │ 模块级代理函数                   │
                └──────────────────────────────────┘

BaseModel — 模型基类

文件位置：src/common/base_model.py

设计目标

为所有业务模型提供统一的通用字段和序列化方法，消除重复定义。

通用字段

python

# src/common/base_model.py
from src.core.database import db
from sqlalchemy import text, func

def now():
    """返回当前时间（北京时间）"""
    from datetime import datetime
    return datetime.now()

class BaseModel(db.Model):
    __abstract__ = True  # 抽象基类，不会在数据库中创建对应的表

    # 主键：自增整数
    id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment="主键ID")
    # 创建人：记录创建者，由业务层通过 get_realname() 设置
    create_user = db.Column(db.String(64), default='', server_default=text("''"), comment="创建人")
    # 创建时间：记录创建时间，创建后不再变化
    create_time = db.Column(db.DateTime, default=now, server_default=func.now(), comment="创建时间")
    # 更新人：记录最后修改者
    update_user = db.Column(db.String(64), default='', server_default=text("''"), comment="更新人")
    # 更新时间：记录最后修改时间，更新时自动刷新
    update_time = db.Column(db.DateTime, default=now, onupdate=now, server_default=func.now(), comment="最后更新时间")
    # 逻辑删除标记：0=正常，1=已删除
    is_delete = db.Column(db.Integer, default=0, server_default=text('0'), index=True, comment="逻辑删除标识：0-正常，1-已删除")

字段	类型	说明
`id`	Integer, PK, AI	主键自增
`create_user`	String(64)	创建人（realname），默认空字符串
`create_time`	DateTime	创建时间，server_default=func.now()
`update_user`	String(64)	更新人（realname），默认空字符串
`update_time`	DateTime	更新时间，onupdate 自动触发
`is_delete`	Integer	软删除标识：0=正常，1=已删除

表注释钩子

python

# src/common/base_model.py
def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    comment = getattr(cls, '__table_comment__', None)
    if not comment:
        return
    if not getattr(cls, '__tablename__', None):
        return
    existing = getattr(cls, '__table_args__', {})
    if isinstance(existing, dict):
        merged = {**existing}
    elif isinstance(existing, tuple) and existing:
        merged = dict(existing[-1]) if isinstance(existing[-1], dict) else {}
    else:
        merged = {}
    if 'comment' not in merged:
        merged['comment'] = comment
        cls.__table_args__ = merged

子类只需声明 __table_comment__ = "岗位表"，基类通过 __init_subclass__ 钩子自动将其写入 __table_args__，无需手动配置 Table(comment=...)。

序列化方法

关键设计点：

敏感字段排除：is_delete、password、salt 不会出现在序列化结果中
时区处理：MySQL 的 DATETIME 读回为 naive UTC，统一转 aware UTC 后再转本地时间
驼峰命名：数据库下划线字段名自动转为前端习惯的驼峰命名

BaseDB — 数据库操作基类

文件位置：src/common/base_db.py

设计目标

提供模型实例的持久化方法，事务边界由中间件统一管理。

python

# src/common/base_db.py
class BaseDB:
    def save(self):
        # 将当前模型实例添加到 SQLAlchemy 会话的持久化上下文中
        db.session.add(self)
        # 立即发送 INSERT SQL 到数据库，获取自增 ID，但不提交事务（事务由中间件统一控制）
        db.session.flush()
        # 返回当前实例，支持链式调用，便于后续操作
        return self

为什么 save() 不 commit？

事务提交由 src/middleware/transaction.py 的 register_transaction 统一管理：响应体 code==0 时 commit，否则 rollback。这样业务代码无需关心事务边界，也避免了部分操作成功、部分失败导致的数据不一致。审计日志等需要独立提交的场景，可使用 commit_independently(obj) 绕过请求事务。

BaseForm — 表单基类

文件位置：src/common/forms/base_form.py

设计目标

所有创建/编辑表单的 WTForms 基类，提供统一的可选 id 字段和前端空字符串容错处理。

完整代码

python

# src/common/forms/base_form.py
# +======================================================================
# | 模块: 表单基类
# | 说明: 统一处理前端传入的空字符串 id → None 等通用校验
# +======================================================================

from flask_wtf import FlaskForm
from src.common.forms.fields import ArrayField


class BaseForm(FlaskForm):
    """
    自定义表单基类，增强对 JSON 数组的支持

    重写 process 方法，确保 ArrayField 能正确处理数组数据。
    """

    def process(self, formdata=None, obj=None, data=None, **kwargs):
        """重写 process 方法，支持 ArrayField 处理数组数据"""
        # ... 处理逻辑

代码解析

继承 FlaskForm：BaseForm 继承自 flask_wtf.FlaskForm（而非 wtforms.Form），获得 CSRF 保护等 Flask 集成能力。

ArrayField 支持：重写 process 方法，确保 ArrayField 能正确处理前端传入的 JSON 数组数据。

id 字段由子类自行定义：BaseForm 本身不定义 id 字段，各业务表单类根据需要自行声明（通常为 IntegerField + Optional()）。

子类示例

python

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

class PositionForm(BaseForm):
    """岗位创建/编辑表单"""
    name = StringField('岗位名称', validators=[...])
    status = IntegerField('岗位状态', validators=[...])
    sort = IntegerField('岗位排序', validators=[...])

PositionForm 继承 BaseForm 后自动获得可选 id 字段。创建时 JSON 请求体无需包含 id：

json

{ "name": "测试岗位", "status": 1, "sort": 0 }

编辑时传入 id：

json

{ "id": 1, "name": "修改后", "status": 1, "sort": 1 }

使用场景

场景	是否传 id	BaseForm 行为
新增	不传	`id = None`，Service 走 `add()` 分支
编辑	传入	`id = 123`，Service 走 `update()` 分支
特定操作	不需要 id	继承 `Form` 而非 `BaseForm`（如 `PositionStatusForm`）

何时继承 BaseForm vs Form

创建/编辑共用的表单：继承 BaseForm，自动获得可选 id
特定操作的表单（如状态更新、排序调整）：直接继承 Form，只定义必要字段

BaseRepository — 仓库基类

文件位置：src/common/base_repository.py

设计目标

封装各模型的数据访问层公共操作，提供统一的 CRUD 接口和软删除约定。

核心方法一览

方法	签名	说明	软删过滤
`get_by_id`	`(obj_id) → Optional[M]`	按 ID 获取单条记录	是
`get_by_ids`	`(ids) → List[M]`	按 ID 列表批量获取	是
`first_by`	`(filters) → Optional[M]`	等值条件获取单条	是
`exists_by`	`(filters, exclude_id) → bool`	唯一性校验	是
`count_by`	`(filters) → int`	等值条件计数	是
`exists`	`(obj_id) → bool`	检查记录是否存在	是
`count_children`	`(parent_id) → int`	统计子记录数	是
`find_page`	`(...) → (items, total, pages)`	统一分页查询	是
`find_list`	`(...) → List[M]`	统一不分页查询	是
`create`	`(**kwargs) → M`	创建记录	—
`update_by_id`	`(obj_id, data) → int`	按 ID 更新	—
`update_status`	`(obj_id, status) → int`	更新状态字段	—
`delete_by_id`	`(obj_id) → bool`	软删除单条	—
`batch_delete`	`(ids) → int`	批量软删除	—
`batch_restore`	`(ids) → int`	批量恢复	—

软删除约定

重要设计约定

等值便捷方法（get_by_id / get_by_ids / get_all / first_by / count_by / exists / exists_by / find_page / find_list）自动追加 is_delete=0 过滤，调用方无需关心
原始出口 filter(*criterion) 不做软删过滤，由调用方自行追加 Model.is_delete == 0，用于 in_ / != / and_ / 区间等复杂条件

数据清洗

python

# src/common/base_service.py
@classmethod
def _clean_form_data(cls, form):
    """
    过滤表单数据

    仅保留模型真实列字段，防止伪字段进入 update_by_id 触发 SQL 错误。
    camelCase 字段名自动转为 snake_case 后与模型列集合比对。
    """
    if form is None:
        return {}
    columns = {c.name for c in cls.repository.model.__table__.columns}
    return {k: v for k, v in form.data.items() if convert_snake_case(k) in columns}

WTForms Schema 可能用 camelCase（如 parentId），而模型列是 snake_case（如 parent_id）。_clean_form_data 在 Service 层统一转换，同时过滤掉模型中不存在的字段，业务模块无需各自处理。

BaseService — 服务基类

文件位置：src/common/base_service.py

设计目标

通过模板方法模式封装通用 CRUD 流程。子类只需声明差异点，即可获得分页、详情、添加、编辑、删除、批量删除、状态切换等标准接口。

差异点声明

差异点	类型	说明
`repository`	`BaseRepository`	必填，数据仓库类
`model`	`type[M]`	必填，ORM 模型类
`repository`	Repository 类	数据访问层类
`form`	Form 类	新增/更新表单校验类
`status_form`	Form 类	状态更新表单校验类（可选）
`model_name`	str	业务对象名称，用于错误提示
`serialize_maps`	dict	枚举显示名映射（可选）
`serialize_maps`	`Dict[str, Dict]`	枚举显示名映射
`file_fields`	`Tuple[str, ...]`	文件字段（add/update 时自动迁移临时文件，serialize 时自动补全 URL）
`rich_text_fields`	`Tuple[str, ...]`	富文本字段（含嵌入图片迁移 + XSS 清洗）

标准接口

方法	说明	返回值
`get_page(params)`	分页查询	`dict`（含 records/total 等分页数据）
`get_list(params)`	不分页查询	`list`（记录列表）
`get_detail(obj_id)`	详情查询	`dict` 或 `None`
`add(data)`	新增记录	`None`（成功无返回，失败 raise BizError）
`update(data)`	编辑记录	`None`（成功无返回，失败 raise BizError）
`remove(ids)`	批量软删除	`int`（删除数量）
`update_status(data)`	状态切换	`None`（成功无返回，失败 raise BizError）

Service 不返回 R 对象

Service 层入参为数据字典、出参为纯数据（或 None），业务失败通过 raise BizError 表达。Router 层负责封装 R.ok() / R.failed() 响应。

钩子方法

钩子	触发时机	默认行为
`before_add(form)`	新增前	空操作，可返回额外字段字典
`after_add(obj)`	新增后	空操作
`before_update(form, obj)`	编辑前	空操作，可返回额外字段字典
`after_update(obj)`	编辑后	空操作
`before_delete(ids)`	删除前	空操作，抛 BizError 中止
`after_delete(count)`	删除后	空操作
`_enrich_page_records(records)`	分页富化	应用 serialize_maps
`_detail_dict(obj)`	详情序列化	model_to_dict + serialize_maps
`_validate_form(data)`	表单校验	调用 form.validate()
`_clean_form_data(form)`	表单数据过滤	保留模型列字段

实例：Position 模块的基类使用

Model（继承 BaseModel + BaseDB）

python

# 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="岗位名称")
    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)

获得的能力：

6 个通用字段（id / create_user / create_time / update_user / update_time / is_delete）
to_dict() 序列化方法
save() 持久化方法
表注释自动写入

Repository（继承 BaseRepository）

python

# 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',
        )

获得的能力：

全部通用 CRUD 方法（classmethod）
软删除自动过滤
camelCase → snake_case 数据清洗

Service（继承 BaseService）

python

# src/modules/system/position/service.py
from src.common.base_service import BaseService
from src.core.exceptions import BizError
from src.modules.system.position.schema import PositionForm, PositionStatusForm
from src.modules.system.position.repository import PositionRepository
from src.modules.system.user.repository import UserRepository


class PositionService(BaseService):
    """岗位服务类"""
    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} 已被用户引用，无法删除")


# 模块级代理函数
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

获得的能力：

8 个标准接口
唯一性自动校验
分页查询自动构建
序列化自动处理

极简 CRUD

对于最简单的模块（如 Level 职级），Repository 和 Service 只需声明差异点，无需编写任何自定义方法，即可获得完整的增删改查能力。

总结

五大基类通过模板方法模式和声明式配置，将通用 CRUD 流程固化为标准实现：

基类	职责	核心能力
`BaseModel`	模型基类	通用字段、to_dict() 序列化
`BaseDB`	持久化基类	save() 方法
`BaseForm`	表单基类	可选 id、空字符串容错
`BaseRepository`	仓库基类	12 个 CRUD 方法、软删除自动过滤
`BaseService`	服务基类	8 个标准接口、钩子方法、声明式配置

子类只需声明差异点（过滤字段、排序规则、唯一性约束等），即可获得企业级的增删改查能力。以岗位模块为例，完整实现只需约 80 行有效代码（不含版权头），其中 Repository 几乎零代码。这种设计使得新模块的开发效率提升 3-5 倍，同时保证了代码风格和行为的一致性。

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

🔄 事务管理

📁 文件处理

🔧 核心工具

🧩 辅助工具

核心基类体系

基类继承关系

BaseModel — 模型基类

设计目标

通用字段

表注释钩子

序列化方法

BaseDB — 数据库操作基类

设计目标

BaseForm — 表单基类

设计目标

完整代码

代码解析

子类示例

使用场景

BaseRepository — 仓库基类

设计目标

核心方法一览

软删除约定

数据清洗

BaseService — 服务基类

设计目标

差异点声明

标准接口

钩子方法

实例：Position 模块的基类使用

Model（继承 BaseModel + BaseDB）

Repository（继承 BaseRepository）

Service（继承 BaseService）

总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

核心基类体系 ​

基类继承关系 ​

BaseModel — 模型基类 ​

设计目标 ​

通用字段 ​

表注释钩子 ​

序列化方法 ​

BaseDB — 数据库操作基类 ​

设计目标 ​

BaseForm — 表单基类 ​

设计目标 ​

完整代码 ​

代码解析 ​

子类示例 ​

使用场景 ​

BaseRepository — 仓库基类 ​

设计目标 ​

核心方法一览 ​

软删除约定 ​

数据清洗 ​

BaseService — 服务基类 ​

设计目标 ​

差异点声明 ​

标准接口 ​

钩子方法 ​

实例：Position 模块的基类使用 ​

Model（继承 BaseModel + BaseDB） ​

Repository（继承 BaseRepository） ​

Service（继承 BaseService） ​

总结 ​

核心基类体系

基类继承关系

BaseModel — 模型基类

设计目标

通用字段

表注释钩子

序列化方法

BaseDB — 数据库操作基类

设计目标

BaseForm — 表单基类

设计目标

完整代码

代码解析

子类示例

使用场景

BaseRepository — 仓库基类

设计目标

核心方法一览

软删除约定

数据清洗

BaseService — 服务基类

设计目标

差异点声明

标准接口

钩子方法

实例：Position 模块的基类使用

Model（继承 BaseModel + BaseDB）

Repository（继承 BaseRepository）

Service（继承 BaseService）

总结