简体中文 English 繁體中文 日本語 Русский
本文为静态镜像,内容以交互版为准 在交互式文档中心打开 →

数据模型层(ORM)

2.9.0 起框架内置声明式数据模型层:继承 Model 并用 Field 声明字段,即可获得自动建表与增删改查能力。模型直接构建在内置存储层之上——SQLite / MySQL / PostgreSQL 由 ErisPulse.storage.backend 配置决定,切换后端无需修改模型代码。

声明模型

from ErisPulse.Core.Bases import Model, Field

class User(Model):
    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0, ge=0, le=150)
    role: str = Field(default="user", choices=["user", "admin"])
    tags: list = Field(default_factory=list)          # JSON 列,读写自动序列化
    bio: str = Field(default="", description={"i18n": "user.bio", "default": "简介"})

字段参数

参数 说明
default 默认值(未提供且非自增 → 必填 NOT NULL)
default_factory 可变默认值工厂(如 list)
primary_key 主键
autoincrement 自增主键(隐含主键,create 后自动回填)
max_length 字符串最大长度(生成 VARCHAR(n),写入校验)
nullable 是否允许 NULL(默认 False)
index 生成普通索引
unique 唯一约束
choices / ge / le 枚举 / 数值范围(写入校验)
description 描述(i18n 字典,与配置类同格式)
column_type 覆写 SQL 列类型定义

建表与 CRUD

await User.create_table()                      # 幂等(IF NOT EXISTS)

user = await User.create(name="Alice", age=20) # 插入(自增主键自动回填)
got = await User.get(id=user.id)               # 等值查询首条

users = await User.where(User.age > 18).order_by("-age").limit(10).all()
first = await User.where(User.name == "Alice").first()
total = await User.count(User.age > 18)

user.age = 21
await user.save()                              # 按主键更新(先约束校验);主键缺失时退化为插入(自增主键同样回填到实例)
await user.delete()                            # 按主键删除

await User.update_all(User.age > 18, role="adult")  # 批量更新
await User.delete_all(User.age > 100)               # 批量删除

查询表达式支持 > >= < <= == !=、in_([...]),以及 &(与)/ |(或)组合:

await User.where((User.age > 18) & User.name.in_(["Alice", "Bob"])).all()

事务

ORM 读写与框架存储层共用同一事务路由:处于 storage.atransaction() 环境事务内时,create / save / delete 与查询自动复用事务连接,随事务统一提交或回滚,不会独立提交。

async with storage.atransaction():
    await User.create(name="Alice")
    ...  # 块内抛异常时,上面的 INSERT 一并回滚

与声明式配置类的关系

模型声明与 ConfigClass(@dataclass + field(metadata=...))共享同一套底层——约束词表、校验器引擎(validate_field_constraints)、类型类别注册表(python_type_category);但类基座有意分立:配置字段是普通值(TOML 往返、一次加载热更新),模型字段是列描述符(类访问 = 查询表达式、逐行实例)。一份声明语法,两个各司其职的基座。

两种声明何时用哪个

维度 配置类 field(metadata=...) 模型字段 Field()
适用场景 模块行为参数(少量、人工可读、需热更新) 业务数据记录(多行、程序读写、需查询)
存储形态 config.toml(注释保留、TOML 往返) 数据库表(自动建表、SQL 方言)
值形态 普通值(dataclass 属性直读) 描述符(类访问 = 列表达式,实例访问 = 行值)
约束声明 metadata={"choices": ..., "min": ..., "max": ...} Field(choices=..., ge=..., le=..., max_length=...)
共享底层 校验器引擎 + 约束词表 + 类型类别注册表(同一套) 同左

经验法则:**"模块怎么运转"用配置类,"用户产生了什么数据"用模型**。

边界与注意事项

自动迁移(阶段二)

create_table() 在表已存在时自动对比现有列与模型字段:新增字段自动执行 ALTER TABLE ADD COLUMN(迁移列剔除 NOT NULL 约束,存量行回填 NULL), 声明了 index=True 的新字段同步建索引。无需手工写迁移脚本。

# v1 上线后模型演进:新增 email / bio 字段
class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0)
    email: str = Field(default="")     # 新增:下次 create_table() 自动 ADD COLUMN
    bio: str = Field(default="")

await User.create_table()              # 幂等:仅迁移新增列

边界:仅支持新增列;主键变更、列类型变更、删列需手工处理(避免破坏性 ALTER 误操作)。

外键(关系映射基础)

foreign_key="表.列" 声明列级外键约束,DDL 生成 REFERENCES 子句:

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")

    content: str = Field(max_length=255)

关系映射(relationship)

在模型类体中以类属性声明 relationship(),方向按外键列声明在哪张表自动判定:

from ErisPulse.Core.Bases import Model, Field, relationship

class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)

    posts = relationship("Post", foreign_key="author")   # 外键在对方表 → has-many

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")
    content: str = Field(max_length=255)

    writer = relationship("User", foreign_key="author")  # 外键在本表 → belongs-to

has-many:实例属性返回查询集,QuerySet 全部链式能力可用, create 自动回填本表主键到对方外键列:

alice = await User.get(name="Alice")

posts = await alice.posts.all()                          # 该用户的全部文章
latest = await alice.posts.order_by("-id").first()
total = await alice.posts.count()
hot = await alice.posts.where(Post.content != "").all()  # 追加对方表字段条件
await alice.posts.delete()                               # 只删该用户的

new_post = await alice.posts.create(content="hi")        # author 自动 = alice.id

belongs-to:直接 await 得到对方实例(无匹配或外键为 NULL 返回 None):

post = await Post.get(id=1)
writer = await post.writer          # User 实例或 None
await writer.posts.count()          # 双向互通

要点: