djangoadmin/djangoadmin-wiki

Fork 0

Files

T

gitadmin 20d8a85989 签入版本

2026-09-16 10:11:44 +08:00

25 KiB

Raw Blame History

核心基类体系

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

基类继承关系

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

┌──────────────────────────────────────────────────────────────────┐
│ SQLAlchemy Base (declarative_base)                               │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────────┐
│ BaseModel (src/core/base_model.py)                               │
│ ─────────────────────────────────────────────────────────────── │
│ 通用字段: id / create_user / create_time / update_user /         │
│ update_time / is_delete                                          │
│ 序列化: to_dict() → 驼峰字段名 + UTC 转本地 + 排除敏感字段       │
│ 子类钩子: __table_comment__ 自动写入 __table_args__              │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                Position(BaseModel)
                ┌──────────────────────────┐
                │ __tablename__            │
                │ name / status / sort     │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseSchema (src/core/base_schema.py)                             │
│ ─────────────────────────────────────────────────────────────── │
│ Pydantic BaseModel 子类                                          │
│ 可选 id 字段: Optional[int]，创建时可省略                        │
│ 空字符串容错: '' / 'null' / 'undefined' → None                   │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                PositionSchema(BaseSchema)
                ┌──────────────────────────┐
                │ name / status / sort     │
                │ 继承可选 id              │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseRepository (src/core/base_repository.py)                     │
│ ─────────────────────────────────────────────────────────────── │
│ 查询: get_by_id / get_by_ids / get_one / all / count             │
│ 分页: paginate → (items, total)                                  │
│ 写操作: create / update / batch_delete                           │
│ 条件: filter (原始) / filter_by (等值+软删)                      │
│ 软删除: 等值方法自动追加 is_delete=0                             │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                PositionRepository(BaseRepository)
                ┌──────────────────────────┐
                │ model = Position         │
                └──────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│ BaseService (src/core/base_service.py)                           │
│ ─────────────────────────────────────────────────────────────── │
│ 查询: get_page / get_options / get_detail                        │
│ 写操作: add / update / delete / batch_delete / update_status     │
│ 差异点: repo_cls / model / page_like_fields / page_eq_fields /   │
│ page_order_by / unique_fields / serialize_maps                   │
│ 钩子: _before_add / _before_update / _before_delete              │
│ _serialize / _serialize_detail                                   │
│ _build_create_fields / _build_update_fields                      │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ▼
                PositionService(BaseService)
                ┌──────────────────────────────────┐
                │ repo_cls = PositionRepository    │
                │ model = Position                 │
                │ page_like_fields = ('name',)     │
                │ page_eq_fields = ('status',)     │
                │ page_order_by = (('sort','asc'),)│
                │ unique_fields = {'name': '...'}  │
                │ position_service = 单例          │
                └──────────────────────────────────┘

base_model — 模型基类

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

设计目标

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

通用字段

class BaseModel(Base):
    __abstract__ = True  # 抽象基类，不会在数据库中创建对应的表
    # 主键：自增整数，所有业务表的唯一标识
    id = Column(Integer, primary_key=True, autoincrement=True, comment="主键ID")
    # 创建人：记录该条记录的创建者姓名，由 Repository 层自动填充
    create_user = Column(String(50), nullable=True, comment="创建人")
    # 创建时间：记录创建时的 UTC 时间，数据库默认自动生成
    create_time = Column(DateTime, default=_UTC_NOW, server_default=func.now(), comment="创建时间")
    # 更新人：记录最后修改该条记录的姓名，由 Repository 层自动填充
    update_user = Column(String(50), nullable=True, comment="更新人")
    # 更新时间：记录最后修改时间，更新时自动刷新，数据库默认自动生成
    update_time = Column(DateTime, default=_UTC_NOW, onupdate=_UTC_NOW, server_default=func.now(), comment="最后更新时间")
    # 逻辑删除标记：0 表示正常记录，1 表示已删除，物理删除不使用，保证数据可恢复；添加索引以提升查询性能，Repository 层默认过滤 is_delete=0
    is_delete = Column(Integer, default=0, server_default=text('0'), index=True, comment="逻辑删除标识：0-正常，1-已删除")

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

表注释钩子

# src/core/base_model.py
# ============================================================
# 子类钩子：自动将 __table_comment__ 写入 __table_args__
# ============================================================
def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    comment = getattr(cls, '__table_comment__', None)
    if not comment:
        return
    # 仅对有 __tablename__ 的具体模型操作（跳过纯抽象中间层）
    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=...)。

序列化方法

# src/core/base_model.py
def to_dict(self) -> Dict[str, Any]:
    """将模型实例转换为字典"""
    result = {}
    for column in self.__table__.columns:
        if column.name in SERIALIZE_EXCLUDE_FIELDS:
            continue
        value = getattr(self, column.name)
        if isinstance(value, datetime):
            if value.tzinfo is None:
                # MySQL DATETIME 无时区信息，读回为 naive UTC，统一按 UTC 转本地
                value = value.replace(tzinfo=timezone.utc)
            value = value.astimezone().strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(value, date):
            value = value.strftime('%Y-%m-%d')
        # 字段名转驼峰
        camel_key = snake_to_camel(column.name)
        result[camel_key] = value
    return result

关键设计点：

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

BaseSchema — 表单基类

文件位置：src/core/base_schema.py

设计目标

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

完整代码

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

from typing import Optional
from pydantic import BaseModel, Field, field_validator

# ============================================================
# 表单基类
# ============================================================
class BaseSchema(BaseModel):
    """
    表单基类

    所有创建/编辑表单应继承此类，自动获得：
        1. id 字段：Optional[int]，前端传空字符串时自动转为 None
    """

    id: Optional[int] = Field(None, description="主键ID")

    @field_validator('id', mode='before')
    @classmethod
    def empty_str_to_none(cls, v):
        """前端 POST 空字符串 '' 时转为 None，避免 Pydantic 整数解析报错"""
        if v == '' or v == 'null' or v == 'undefined':
            return None
        return v

代码解析

可选 id 字段：

id: Optional[int] = Field(None, description="主键ID")

id 默认为 None。创建记录时前端不传 id，Pydantic 自动赋默认值；编辑记录时前端传入 id，Service 层据此判断是新增还是更新。

空字符串容错：

@field_validator('id', mode='before')
@classmethod
def empty_str_to_none(cls, v):
    if v == '' or v == 'null' or v == 'undefined':
        return None
    return v

前端表单在未填写 id 时，可能提交空字符串 ''、字面量 'null' 或 'undefined'。Pydantic 无法将这些值解析为 int，会抛出校验错误。mode='before' 确保此验证器在 Pydantic 类型转换之前执行，将这些边界值统一转为 None。

子类示例

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

# ============================================================
# 岗位表单类
# ============================================================
class PositionForm(BaseSchema):
    """岗位创建/编辑表单"""
    name: str = Field(..., min_length=1, max_length=150, description="岗位名称")
    status: int = Field(..., ge=1, le=2, description="岗位状态：1-在用 2-停用")
    sort: int = Field(..., ge=0, le=99999, description="岗位排序")

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

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

编辑时传入 id：

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

使用场景

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

:::tip 何时继承 BaseSchema vs BaseModel

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

BaseRepository — 仓库基类

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

设计目标

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

核心方法一览

方法	签名	说明	软删过滤
`get_by_id`	`(obj_id) → Optional[M]`	按 ID 获取单条记录	是
`get_by_ids`	`(ids) → list`	按 ID 列表批量获取	是
`get_one`	`(**filters) → Optional[M]`	等值条件获取单条	是
`all`	`(**filters) → List[M]`	等值条件获取所有	是
`count`	`(**filters) → int`	等值条件计数	是
`exists`	`(obj_id) → bool`	检查记录是否存在	是
`exists_by_field`	`(field, value, exclude_id) → bool`	唯一性校验	是
`paginate`	`(page_no, page_size, query) → (items, total)`	分页查询	是
`filter`	`(*criterion) → Query`	原始条件查询	否
`filter_by`	`(**kwargs) → Query`	等值条件查询	是
`create`	`(**fields) → M`	创建记录	—
`update`	`(obj_id, data) → int`	按 ID 更新	是
`batch_delete`	`(ids_str) → int`	按 ID 软删除	—

软删除约定

:::warning 重要设计约定

等值便捷方法（get_by_id / get_one / all / count / filter_by / exists / exists_by_field / paginate）自动追加 is_delete=0 过滤，调用方无需关心
原始出口 filter(*criterion) 不做软删过滤，由调用方自行追加 Model.is_delete == 0，用于 in_ / != / and_ / 区间等复杂条件 :::

数据清洗

# src/core/base_repository.py
def _clean_model_data(self, data: dict) -> dict:
    """清洗入库数据：camelCase 键转 snake_case 并仅保留模型真实列字段

    Pydantic schema 可能用 camelCase（如 parentId）而模型列是 snake_case
    （parent_id），直传 Query.update / model(**fields) 会触发 CompileError /
    TypeError。统一在数据访问层转换，业务模块无需各自处理。
    """
    if not data:
        return {}
    columns = {c.name for c in self.model.__table__.columns}
    return {
        _camel_to_snake(k): v
        for k, v in data.items()
        if _camel_to_snake(k) in columns
    }

Pydantic Schema 可能用 camelCase（如 parentId），而模型列是 snake_case（如 parent_id）。_clean_model_data 在数据访问层统一转换，同时过滤掉模型中不存在的字段，业务模块无需各自处理。_camel_to_snake 是 BaseRepository 的静态方法。

BaseService — 服务基类

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

设计目标

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

差异点声明

差异点	类型	说明
`repo_cls`	`Type[BaseRepository]`	必填，数据仓库类（不是实例）
`model`	`type[M]`	必填，ORM 模型类
`page_like_fields`	`Tuple[str, ...]`	分页模糊查询字段（取同名查询参数，`LIKE %值%`）
`page_eq_fields`	`Tuple[str, ...]`	分页等值查询字段（取同名查询参数，`=值`）
`page_order_by`	`Tuple[Tuple[str, str], ...]`	分页排序规则，默认 `(('id', 'desc'),)`
`unique_fields`	`Dict[str, str]`	唯一性校验字段及提示，如 `{'name': '岗位名称不能重复'}`
`serialize_maps`	`Dict[str, Dict]`	枚举显示名映射。值为 `str` 时从数据字典取 `{value: label}`，值为 `dict` 时直接按值取名称
`serialize_extra`	`Callable`	列表/详情共用序列化钩子，返回需合并进字典的额外字段
`detail_serialize`	`Callable`	详情专用序列化钩子，仅在 `get_detail` 时合并
`file_fields`	`Tuple[str, ...]`	文件字段（add/update 时自动迁移临时文件，serialize 时自动补全 URL）
`rich_text_fields`	`Tuple[str, ...]`	富文本字段（含嵌入图片迁移 + XSS 清洗）
`file_dir`	`str`	文件存储子目录名，默认从表名去掉 `DB_PREFIX` 推导

标准接口

方法	说明	返回值
`get_page(handler)`	分页查询	`R.page(data, total, current, size)`
`get_options(handler)`	下拉列表查询	`R.ok(data=[...])`
`get_detail(obj_id)`	详情查询	`dict` 或 `None`
`add(handler)`	新增记录	`R.ok(msg='添加成功')`
`update(handler)`	编辑记录	`R.ok(msg='更新成功')`
`delete(handler, ids)`	删除记录	`R.ok(msg='本次共删除N条数据')`
`batch_delete(handler)`	批量删除	`R.ok(...)`
`update_status(handler)`	状态切换	`R.ok(msg='设置成功')`

钩子方法

钩子	触发时机	默认行为
`_before_add(handler, data)`	新增前	空操作
`_before_update(handler, data)`	编辑前	空操作
`_before_delete(ids)`	删除前	返回 None（放行）
`_build_page_query(handler)`	分页查询构建	按声明字段过滤 + 排序
`_serialize(item)`	列表行序列化	`to_dict()` + 枚举映射 + 文件 URL
`_serialize_detail(obj)`	详情序列化	同列表 + `detail_serialize`
`_build_create_fields(data)`	新增字段组装	camelCase→snake_case + 过滤非模型字段
`_build_update_fields(data)`	更新字段组装	同上 + 排除 None 值

删除流程

Service 层的 delete() 和 batch_delete() 方法内部调用 repo.batch_delete(ids) 完成软删除：

# src/core/base_service.py
async def delete(self, handler, ids) -> R:
    try:
        id_list = list(dict.fromkeys(parse_id_list(str(ids))))
        if not id_list:
            return R.failed(handler, "记录ID不存在")
        if len(self.repo.get_by_ids(id_list)) != len(id_list):
            return R.failed(handler, "记录不存在")
        err = self._before_delete(ids)
        if err:
            return R.failed(handler, err)
        count = self.repo.batch_delete(ids)
        return R.ok(handler, msg="本次共删除{0}条数据".format(count))
    finally:
        self.repo.close()

:::tip 设计要点 Repository 层的 batch_delete 只返回删除条数（数据原语），Service 层负责 R 响应封装和文案统一。删除前先校验全部记录存在，避免"部分软删已提交却整体报失败"导致的数据不一致。 :::

实例：Position 模块的基类使用

Model（继承 BaseModel）

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

from config.database import DB_PREFIX
from core.base_model import BaseModel


class Position(BaseModel):
    """岗位模型。"""
    __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="岗位状态")
    # 岗位排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="岗位排序")

获得的能力：

6 个通用字段（id / create_user / create_time / update_user / update_time / is_delete）
to_dict() 序列化方法（自动排除 password/salt/is_delete，驼峰命名）
__table_comment__ 自动写入 __table_args__

Repository（继承 BaseRepository）

# src/modules/position/repository.py
"""岗位数据访问。"""

from core.base_repository import BaseRepository
from modules.position.models import Position


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

获得的能力：

全部通用 CRUD 方法（get_by_id / get_by_ids / get_one / all / count / exists / exists_by_field / paginate / filter / filter_by / create / update / batch_delete）
软删除自动过滤（等值方法自动追加 is_delete=0）
camelCase → snake_case 数据清洗（_clean_model_data）

Service（继承 BaseService）

# src/modules/position/service.py
"""岗位业务逻辑。"""

from core.base_service import BaseService
from modules.position.models import Position
from modules.position.repository import PositionRepository
from modules.position.schemas import PositionSchema, PositionStatusSchema


class PositionService(BaseService):
    """岗位服务。"""
    # 基础声明
    model = Position
    repo_cls = PositionRepository
    # 差异点声明
    create_schema = PositionSchema
    update_schema = PositionSchema
    status_schema = PositionStatusSchema
    page_like_fields = ("name",)
    page_eq_fields = ("status",)
    page_order_by = (("sort", "asc"),)
    options_filter = {"status": 1}


# 模块级单例
position_service = PositionService()

获得的能力：

8 个标准接口（get_page / get_options / get_detail / add / update / delete / batch_delete / update_status）
唯一性自动校验
分页查询自动构建
序列化自动处理

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

总结

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

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

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

25 KiB Raw Blame History Unescape Escape

核心基类体系

基类继承关系

base_model — 模型基类

设计目标

通用字段

表注释钩子

序列化方法

BaseSchema — 表单基类

设计目标

完整代码

代码解析

子类示例

使用场景

BaseRepository — 仓库基类

设计目标

核心方法一览

软删除约定

数据清洗

BaseService — 服务基类

设计目标

差异点声明

标准接口

钩子方法

删除流程

实例：Position 模块的基类使用

Model（继承 BaseModel）

Repository（继承 BaseRepository）

Service（继承 BaseService）

总结

25 KiB

Raw Blame History