Python 框架指南 · 第 05 章

ORM 建模

本章你将学会 - 用 DeclarativeBase + Mapped + mappedcolumn 三件套声明模型 - 主键 / 自增 / 可空 / 默认值 / 唯一 / 索引的每一种写法 - 外键 Forei…

约 45 分钟17,818 字34 个代码示例

本章你将学会

用 DeclarativeBase + Mapped + mapped_column 三件套声明模型

主键 / 自增 / 可空 / 默认值 / 唯一 / 索引的每一种写法

外键 ForeignKey 与 ondelete 删除策略

一对多、多对多：relationship、back_populates、secondary

用 Mixin 复用公共字段，用 TYPE_CHECKING 防循环导入

JSONB 等 PG 特有类型，以及 create_all 的真实定位

上一章我们把"路"修好了：引擎管连接池，会话管事务。这一章开始"盖房子"——用 Python 类把表结构写出来，SQLAlchemy 负责翻译成 CREATE TABLE。本章自始至终围绕一个内容平台的小案例：栏目（Channel）、文章（Article）、标签（Tag）、文章版本（ArticleVersion），它们刚好把常用的建模姿势全用上了。

1. 心智模型：类是图纸，实例是砖

ORM 建模只有一句话要记：

类 ↔ 表；类属性 ↔ 列；类实例 ↔ 一行数据。

class Article(Base):        # 对应表 article
    title: Mapped[str]      # 对应列 title
    ...

# 查出来的每个对象，就是表里的一行
article = Article(title="第一篇文章")   # 还没入库，只是图纸上的规划

打个比方：模型类是图纸，Base.metadata 是市政局的档案柜——每定义一个模型类，SQLAlchemy 就自动把这张图纸登记进档案柜。后面不管是一键建表（create_all）还是跑迁移（第 07 章的 Alembic），都是照着档案柜里的图纸开工。图纸没登记，就没人为你施工。

2. DeclarativeBase：所有模型的"户籍管理处"

作用：所有模型类的共同祖先。继承它，SQLAlchemy 才知道"这是一个表模型"，才会把表登记进 metadata 档案柜。

签名（无参数，按继承使用）：

class DeclarativeBase:
    # 可在子类中定制的类属性：metadata、registry、type_annotation_map

最小示例：

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

项目里的第一个模型文件，一般就长这样。需要定制时再传类属性，比如全局类型映射：

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention={   # 统一约束命名，生产环境推荐
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
    })

常见坑：

一个项目只建一个 Base。Base 可以建多个，但每个 Base 各有一个独立档案柜（metadata），create_all 和 Alembic 都只认自己那个柜子——模型挂错了户口，表就"失踪"了。
模型必须被 import 过才存在于档案柜。定义了 Article 类但没被任何代码导入，create_all 根本不知道它存在（第 14 节细说）。
普通的工具基类不要继承 Base。只有"确实要映射成表"的类才继承；复用公共字段用 Mixin（第 12 节），Mixin 自己是普通类。

3. `tablename`：给表起名字

作用：指定模型对应的表名。不写也行——SQLAlchemy 会拿类名当表名，但显式控制更稳。

签名（declarative 指令式属性）：

__tablename__: str                    # 手动指定
# 或用 declared_attr.directive 做自动生成

最小示例——手动写：

class Channel(Base):
    __tablename__ = "channel"

自动生成（推荐做法，全项目统一风格）：

from sqlalchemy.orm import DeclarativeBase, declared_attr

class Base(DeclarativeBase):
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower()   # Channel -> "channel"

写成 declared_attr.directive 的意思是："这是个给 declarative 机制看的指令属性，不是要映射成列的属性"。子类继承时它都会重新执行一次，各自算各自的表名。

常见坑：

自动小写对 ArticleVersion 这种驼峰名会得到 articleversion（没有下划线）。嫌丑就在子类里手动覆盖：__tablename__ = "article_version"，覆盖优先级高于自动生成。
别用数据库保留字当表名：user、order 这类词在 PG 里是关键字，直接建表会翻车，起名 users、orders 绕开。
表名会出现在 ForeignKey("表名.列名") 的字符串里，改表名时记得全局搜一遍这些字符串引用。

4. `Mapped` + `mapped_column`：类型标注驱动的列定义

作用：这是 SQLAlchemy 2.0 的核心建模姿势。Mapped[类型] 注解决定列的"基础类型和可空性"，mapped_column() 补充细节（长度、唯一、索引、默认值……）。相当于看身份证定工种，再按岗位细则派活。

签名（手册式精简版，只列高频参数）：

mapped_column(
    类型 或 ForeignKey(...),   # 第一个位置参数：显式指定列类型或外键
    *,
    primary_key=False,          # 是否主键
    nullable=None,             # 是否可空（默认由注解推断）
    default=None,               # Python 端默认值（flush 时填充）
    server_default=None,        # 数据库端默认值（写进 DDL 的 DEFAULT）
    onupdate=None,              # UPDATE 时自动刷新的值
    index=False,                # 建普通索引
    unique=False,               # 建唯一约束
)

最小示例——三种写法，从简到繁：

class Tag(Base):
    # 写法一：只要注解，零配置（str -> String，非空）
    name: Mapped[str] = mapped_column(String(50), unique=True)

    # 写法二：注解 + 空 mapped_column()，纯占位，将来加参数方便
    note: Mapped[str] = mapped_column()

    # 写法三：可空列——Optional 注解自动推导 nullable=True
    remark: Mapped[str | None] = mapped_column(String(200))

类型注解和列类型的对应关系（默认映射表）：

Python 类型注解	默认列类型
`Mapped[int]`	`Integer`
`Mapped[float]`	`Float`
`Mapped[bool]`	`Boolean`
`Mapped[str]`	`String`（无长度限制）
`Mapped[bytes]`	`LargeBinary`
`Mapped[datetime]`	`DateTime`
`Mapped[date]` / `Mapped[time]`	`Date` / `Time`
`Mapped[timedelta]`	`Interval`
`Mapped[Decimal]`	`Numeric`
`Mapped[uuid.UUID]`	`Uuid`
`Mapped[dict]` / `Mapped[list]`	`JSON`

常见坑：

Mapped[str] 默认映射成不带长度的 String：PG、SQLite 无所谓，MySQL 会直接报错（VARCHAR 必须给长度）。跨库项目养成习惯：字符串列显式写 String(50)。
可空性由注解推断，别两边打架：Mapped[str | None] 自动 nullable=True；这时再手写 nullable=False 是自相矛盾，以显式参数为准，但读代码的人会懵。一个原则：注解管可空，参数管细节。
注解必须真的写在类型里：Mapped[str] 和 Mapped[str | None] 是两种列，写错一个 | None，数据库约束就完全变了。
想让自定义类型也自动映射？在 DeclarativeBase 子类上传 type_annotation_map 扩展映射表，本章不展开。

5. 主键与自增

作用：primary_key=True 声明主键；整型主键默认自增，插入时不用手动给值。

签名：

mapped_column(Identity(*, start=None, increment=None, always=False),
              primary_key=True, autoincrement="auto")

最小示例：

class Channel(Base):
    # 最常见：整型自增主键（整型主键默认自增，不用额外参数）
    id: Mapped[int] = mapped_column(primary_key=True)

# PG / Oracle 风格：标准 IDENTITY 子句（与上面二选一）
# id: Mapped[int] = mapped_column(Identity(), primary_key=True)

Identity() 生成的是 GENERATED BY DEFAULT AS IDENTITY——PG 10+、Oracle、MariaDB 10.3+ 都认；SQLite 这类不支持的库会安静地忽略它，退回普通的整型主键自增（本章 demo 在 aiosqlite 上就是这么跑的，行为完全正常）。

常见坑：

整型单列主键默认就是自增（autoincrement="auto"），不写 Identity() 也行。写它的意义是"声明式地表达我要数据库端自增"，在 PG 上更标准。
autoincrement=False 可以关掉自增——比如 id 由外部系统发号时。
想用 UUID 主键：Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)（Python 端生成），或 PG 13+ 上 server_default=func.gen_random_uuid()（数据库端生成）。
复合主键（多列联合主键）见第 10 节，写法不一样。

6. 可空与默认值：`default` 和 `server_default` 是两套机制

作用：给列配默认值。名字看着像兄弟，其实是两家人：

default=...：Python 端默认值。ORM 在 flush 生成 INSERT 语句时替你补上，不会出现在 DDL 里。
server_default=...：数据库端默认值。翻译成 CREATE TABLE 里的 DEFAULT ... 子句，谁来插行都生效。

打个比方：default 是发货员（ORM）发货时顺手塞进包裹的赠品——绕过这个发货员（直接用 psql 手写 INSERT）就没有了；server_default 是写进仓库规章制度的赠品规则——不管谁来发货，仓库都照章办事。

签名：

mapped_column(..., default=Python值或可调用对象,
              server_default=SQL表达式或字符串,
              onupdate=Python值或SQL表达式)

最小示例：

from datetime import datetime
from sqlalchemy import DateTime, String, Text, func

class Article(Base):
    # Python 端默认值：默认空字符串
    content: Mapped[str] = mapped_column(Text, default="")

    # 可空列：不写摘要就不写，数据库里存 NULL
    summary: Mapped[str | None] = mapped_column(String(500))

    # 数据库端默认值：DDL 里会有 DEFAULT CURRENT_TIMESTAMP
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    # 版本号：Python 端默认 1
    version_no: Mapped[int] = mapped_column(default=1)

onupdate 是 default 的"更新版"：ORM 执行 UPDATE 时自动把列刷新成新值（常配 func.now() 做 updated_at，第 12 节的 TimestampMixin 就是这么写的）。

常见坑：

类型限制：server_default 只接受 SQL 表达式（func.now()、text("''")）或字符串字面量，直接塞 Python 对象（如 datetime.now()）会报错。Python 对象是给 default= 用的。
传可调用对象时别加括号：default=datetime.now 传的是函数本身，每次插入时才调用、各自拿时间；default=datetime.now() 带括号传的是"模块加载那一刻"的固定结果——所有行共用一个时间戳，新手重灾区。
改 default 不影响已有数据：它只作用于之后的 INSERT。想让老数据也补上默认值，得跑一次 UPDATE（第 06 章）。
onupdate 只在 ORM 更新时触发。别人直接 UPDATE ... WHERE ... 的裸 SQL 不会刷新它，别指望它 100% 可靠。

7. 唯一与索引：`unique`、`index`、`__table_args__`

作用：给列加约束或索引。unique=True 唯一约束（数据库层面拦截重复），index=True 普通索引（加速查询，不约束数据）。

签名：

mapped_column(..., unique=False, index=False)

# 类级别：多列组合约束
__table_args__ = (
    UniqueConstraint("列1", "列2", name="约束名"),
    Index("索引名", "列1", "列2"),
)

最小示例：

from sqlalchemy import Index, String, UniqueConstraint

class Tag(Base):
    name: Mapped[str] = mapped_column(String(50), unique=True)   # 单列唯一

class ArticleVersion(Base):
    __tablename__ = "article_version"
    # 组合唯一：同一篇文章里，版本号不能重复
    __table_args__ = (
        UniqueConstraint("article_id", "version_no", name="uq_article_version_no"),
        Index("ix_version_article", "article_id"),   # 显式命名的普通索引
    )
    article_id: Mapped[int] = mapped_column(ForeignKey("article.id"))
    version_no: Mapped[int] = mapped_column(default=1)

建出来的 DDL 长这样（demo 实测）：

CONSTRAINT uq_article_version_no UNIQUE (article_id, version_no)

常见坑：

外键列在 PG 上不会自动建索引（MySQL 会）。按外键查询是常态，建议外键列加 index=True。
index=True 自动生成的索引名各库有各的规则，生产环境想统一管理，用 __table_args__ 里的 Index("ix_名字", "列") 显式命名。
唯一约束撞上可空列：PG 里 NULL 不参与唯一比较，多行 NULL 是合法的。想"最多一个 NULL"得用部分索引（Index(..., postgresql_where=...)），进阶话题，先知道有这回事。
唯一约束在插入重复时抛 IntegrityError，这不是 bug，是约束在干活。怎么优雅地接住它，第 09 章统一异常处理见。

8. `ForeignKey` 与 `ondelete`：外键及其"应急预案"

作用：ForeignKey("表.列") 声明外键，把两张表焊在一起；ondelete 指定父行被删除时，子行怎么办。

签名：

ForeignKey("表名.列名", ondelete=None)   # ondelete: CASCADE / RESTRICT / SET NULL / SET DEFAULT

最小示例：

class Article(Base):
    # 文章必须挂在一个栏目下；栏目还有文章时，栏目不许删（RESTRICT 拦截）
    channel_id: Mapped[int] = mapped_column(
        ForeignKey("channel.id", ondelete="RESTRICT")
    )

class ArticleVersion(Base):
    __tablename__ = "article_version"
    # 文章删了，它的历史版本没存在意义，跟着删
    article_id: Mapped[int] = mapped_column(
        ForeignKey("article.id", ondelete="CASCADE")
    )

四种删除策略，用人话翻译：

策略	含义	适用场景
`CASCADE`	父删，子陪葬	版本记录、日志这种附属数据
`RESTRICT`	有子在，父别想走	栏目下还有文章 → 拦住删除
`SET NULL`	父删，子断联（列必须可空）	可有可无的关联
`SET DEFAULT`	父删，子回默认值	少见

常见坑：

参数是字符串 "表名.列名"，不是 Python 类或属性——因为定义 Article 时，channel 表可能还没被声明。字符串引用让 SQLAlchemy 事后解析。
ondelete 是数据库端行为，生效前提是数据库真的执行外键约束：PG 默认执行；SQLite 默认不执行（要开 PRAGMA foreign_keys=ON），demo 里不深究，生产上 PG 无此烦恼。
还有另一套删除语义在 ORM 层：relationship(cascade=...)（下一节）。新手最容易混：ondelete 管数据库，relationship 的 cascade 管 ORM 内存操作。生产建议两层配套：外键 ondelete="CASCADE" + relationship passive_deletes=True，让数据库干脏活，ORM 少发 SQL。

9. 一对多：`relationship` + `back_populates`

作用：relationship 在两个模型之间架一座内存中的桥——让你用 article.channel、channel.articles 这种属性访问代替手写 JOIN。注意：它只声明对象关系，不产生任何列。建表时数据库只认 mapped_column。

签名（高频参数）：

relationship(
    "目标类名",              # 也可省略，由注解推断
    *,
    back_populates=None,     # 对方模型里"回指"的属性名
    secondary=None,          # 多对多专用，见第 10 节
    cascade=None,            # ORM 级联："all, delete-orphan" 等
    lazy="select",           # 加载策略，第 06 章细讲
    passive_deletes=False,   # 把删除交给数据库的 ondelete
)

最小示例——栏目与文章，一对多（多对一是同一个关系的两面）：

from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Channel(Base):
    ...
    # 一个栏目 → 多篇文章：list["Article"]
    articles: Mapped[list["Article"]] = relationship(back_populates="channel")

class Article(Base):
    ...
    channel_id: Mapped[int] = mapped_column(ForeignKey("channel.id", ondelete="RESTRICT"))
    # 一篇文章 → 一个栏目：单对象
    channel: Mapped["Channel"] = relationship(back_populates="articles")

用起来是什么体验：

article = Article(title="第一篇", channel=Channel(name="后端"))
#                          ↑ 给了对象，channel_id 由 ORM 在 flush 时自动补上

channel.articles        # 直接拿到这个栏目下的所有文章对象

back_populates 是双向对讲机：两边必须互相指认（"channel" ↔ "articles"），任何一边改了，另一边内存里同步更新。

常见坑：

注解方向别写反："多"的一方拿单对象（Mapped["Channel"]），"一"的一方拿列表（Mapped[list["Article"]]）。写反了语义就崩了。
引号别省：Mapped[list["Article"]] 里目标类是字符串（前向引用）；类定义有先后，字符串让 SQLAlchemy 事后解析。
back_populates 指的属性名必须真实存在且互相指认，拼错会在启动时直接报 mapper 配置错误——这反而是好事，早失败早发现。
老代码里常见的 backref=（单向声明自动生成反向）是隐式风格，2.0 官方更推荐 back_populates 显式双向，两边一眼看清。
异步会话里不能懒加载：默认 lazy="select" 意味着访问属性时才发 SQL，而 asyncio 里这个隐式 IO 会抛 MissingGreenlet。查询时用 selectinload / joinedload 预加载，第 06 章展开。

10. 多对多：`secondary` 关联表与复合主键

作用：文章和标签是多对多——一篇文章多个标签，一个标签贴多篇文章。关系数据库表达不了"直接多对多"，必须拆出一个中间表，把一个 M:N 拆成两个 1:N。

中间表用 Core 风格的 Table 定义（它没有自己的业务行为，不需要映射成类）：

签名：

Table(
    "表名",
    Base.metadata,        # 挂到 Base 的档案柜
    Column("article_id", ForeignKey("article.id"), primary_key=True),
    Column("tag_id",      ForeignKey("tag.id"),      primary_key=True),
)
# 两列同时 primary_key=True → 复合主键：(article_id, tag_id) 组合唯一

最小示例——完整的"文章 × 标签"：

from sqlalchemy import Column, ForeignKey, String, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship

article_tag = Table(
    "article_tag",
    Base.metadata,
    Column("article_id", ForeignKey("article.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True),
)

class Article(Base):
    ...
    tags: Mapped[list["Tag"]] = relationship(
        secondary=article_tag, back_populates="articles"
    )

class Tag(Base):
    ...
    articles: Mapped[list["Article"]] = relationship(
        secondary=article_tag, back_populates="tags"
    )

中间表像一张群聊成员表：article_tag 每一行就是"某文章加入了某标签群"。建表 DDL（demo 实测）：

CREATE TABLE article_tag (
    article_id INTEGER NOT NULL,
    tag_id INTEGER NOT NULL,
    PRIMARY KEY (article_id, tag_id),          -- 复合主键
    FOREIGN KEY(article_id) REFERENCES article (id) ON DELETE CASCADE,
    FOREIGN KEY(tag_id) REFERENCES tag (id) ON DELETE CASCADE
)

用起来：

article.tags = [Tag(name="Python"), Tag(name="SQLAlchemy")]
await session.commit()
# ORM 自动往 article_tag 里写两行，不用你管

常见坑：

关联表行由 ORM 自动维护：给 article.tags 赋列表，flush 时自动 INSERT 中间表行；从列表里移除，自动 DELETE。别自己手动往 article_tag 插数据，两条管理路线会打架。
复合主键的写法就是多个 Column(..., primary_key=True)。它天然保证"同一对 (article_id, tag_id) 只出现一次"，重复贴标签会撞唯一约束。
什么时候中间表要升级成模型类？当关联行自己要带字段时（比如"什么时候打的标签"）。这时放弃 secondary，把中间表定义成普通模型类，两边各写一个一对多——这叫 association object 模式。判据一句话：关联本身有没有属性。
中间表也记得挂 Base.metadata，挂了才会被 create_all 建出来。

11. 模型拆文件时：`TYPE_CHECKING` 防循环导入

模型多了一个文件塞不下，自然会按"一表一文件"拆开。问题来了：article.py 要用 Channel，channel.py 要用 Article——你 import 我，我 import 你，Python 直接抛 ImportError（循环导入）。

解法是注解用字符串、导入放条件块：

# model/article.py
from typing import TYPE_CHECKING

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from model.association import article_tag
from model.base import Base, IDMixin, TimestampMixin

if TYPE_CHECKING:
    # 这个块只在类型检查器（mypy / IDE）里执行，运行时不会导入
    from model.channel import Channel
    from model.tag import Tag

class Article(Base, IDMixin, TimestampMixin):
    title: Mapped[str] = mapped_column(String(200), index=True)
    channel_id: Mapped[int] = mapped_column(ForeignKey("channel.id", ondelete="RESTRICT"))

    # 引号字符串 = 前向引用，SQLAlchemy 运行时自己解析，不需要真 import
    channel: Mapped["Channel"] = relationship(back_populates="articles")
    tags: Mapped[list["Tag"]] = relationship(secondary=article_tag, back_populates="articles")

打个比方：if TYPE_CHECKING: 块相当于只在编辑器里递名片——IDE 拿着名片能帮你补全、查类型；运行时大家互不串门，循环导入自然解除。SQLAlchemy 解析关系靠的是字符串名，不靠 import。

常见坑：

只需要类型出现在注解里的，放 TYPE_CHECKING；运行时真要用类本身的（比如继承、secondary= 传表对象），必须正常导入。secondary=article_tag 传的是表对象，所以要真导入。
字符串前向引用别忘引号：Mapped["Channel"]，不带引号运行时就找不到名字了。
建表入口必须把所有模型 import 一遍（哪怕只是 # noqa: F401 占位），否则档案柜缺图纸，create_all 建不全表。第 14 节再强调。

12. Mixin：公共字段的"预制件"

作用：id、created_at、updated_at 这种字段每张表都要，每张表抄一遍就是重复劳动。Mixin 把它们做成预制墙板——浇筑到每栋楼（每个模型）里各自成型，改一处，全楼受益。

最小示例：

# model/base.py
from datetime import datetime

from sqlalchemy import DateTime, Identity, func
from sqlalchemy.orm import DeclarativeBase, Mapped, declared_attr, mapped_column

class Base(DeclarativeBase):
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

class IDMixin:
    """自增主键，全模型通用。"""
    id: Mapped[int] = mapped_column(Identity(), primary_key=True)

class TimestampMixin:
    """创建/更新时间。"""
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

继承时多继承组合即可：

class Article(Base, IDMixin, TimestampMixin):   # id + created_at + updated_at 全有了
    title: Mapped[str] = mapped_column(String(200))
    ...

class Tag(Base, IDMixin):                        # 轻量表，只要 id
    name: Mapped[str] = mapped_column(String(50), unique=True)

常见坑：

Mixin 是普通类，不继承 Base——它是字段模板，不是表模型。只有最终组合出的子类才登记进档案柜。
SQLAlchemy 2.0 允许把 mapped_column 直接写在 Mixin 上，每个子类会自动拿到独立的列副本（不是共享同一个 Column 对象）。
需要按子类动态计算的东西（比如 __tablename__、引用别的类的 relationship）才需要 declared_attr 包一层——第 3 节的 __tablename__ 就是这么写的。
TimestampMixin 里的 server_default=func.now() 是数据库端默认值（DDL 里有 DEFAULT CURRENT_TIMESTAMP），onupdate=func.now() 是 ORM 更新时刷新——两者的边界在第 6 节讲过了。

13. PG 特有类型：以 JSONB 为代表

作用：PG 有一批"别人家没有"的类型，最常用的是 JSONB（二进制存储的 JSON，可索引、可用运算符查询）。存半结构化数据（文章的 meta 信息）时比普通 JSON 强一档。

最小示例：

from sqlalchemy.dialects.postgresql import JSONB, ARRAY
from sqlalchemy import DateTime, String

class Article(Base):
    # PG 原生 JSONB
    meta: Mapped[dict] = mapped_column(JSONB)

    # 带时区的时间戳（PG 的 TIMESTAMPTZ）
    published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # PG 原生数组类型
    keywords: Mapped[list[str]] = mapped_column(ARRAY(String(30)))

跨库兼容技巧——with_variant：demo 要在 SQLite 上跑，但生产用 PG，怎么办？让类型按方言变形：

# PG 上用 JSONB，其它库退化为通用 JSON
meta: Mapped[dict] = mapped_column(JSON().with_variant(JSONB(), "postgresql"))

意思是"默认按 JSON 处理，遇到 postgresql 方言就换成 JSONB"。一行代码，两边通吃。

常见坑：

JSON 和 JSONB 的区别：JSON 存文本原样（快进慢出），JSONB 存解析后的二进制（稍慢进，可索引可查询）。生产基本无脑 JSONB。
ORM 不追踪 JSON 内部变更：article.meta["views"] += 1 之后直接 commit，数据库可能纹丝不动。要么整体重新赋值 article.meta = {...}，要么用 flag_modified(article, "meta") 标脏。
这些类型从 sqlalchemy.dialects.postgresql 导入，换数据库时记得同步改——除非用了上面的 with_variant。

14. `create_all`：一键建表，但它不是迁移工具

作用：扫一遍 Base.metadata 档案柜里登记的所有表，不存在的建出来。开发期的效率神器。

签名：

Base.metadata.create_all(bind, checkfirst=True)   # checkfirst=True：已存在的表跳过

异步姿势（第 04 章的引擎直接接上）：

import asyncio
from model.base import Base
import model.article   # noqa: F401  ← 只 import 不到位，表就进不了档案柜！
import model.tag       # noqa: F401
from sqlalchemy.ext.asyncio import create_async_engine

async def init_db() -> None:
    engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost/db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await engine.dispose()

asyncio.run(init_db())

run_sync 是异步引擎的固定用法：create_all 本身是同步 API，包在 run_sync 里让它在连接上安全执行。

它的定位，必须说清楚：

create_all 是活动板房——快速搭起能用的表结构，适合开发、原型、测试环境。
它只建新表，绝不改旧表：checkfirst=True 的意思是"存在就跳过"，不是"对比差异后更新"。你给 Article 加了一列，create_all 看一眼：表在，跳过。新列永远不会出现。
生产环境的表结构演进要靠 Alembic（第 07 章）：它像一支带施工日志的正规施工队——每次改结构生成一个版本化的迁移脚本，升级、回滚、多环境同步都有据可查。

所以本章 demo 用 create_all 快速起跑，第 07 章马上换正规军。

常见坑：

最经典的坑：建了模型忘了 import，create_all 建出的库里就是少那几张表，而且不报错。
在异步引擎上直接 await Base.metadata.create_all(...) 会失败——它是同步函数，必须走 conn.run_sync(...)。
每次改了模型列结构，指望重跑 create_all 生效？不会生效（见上）。开发期土办法是删库重建，正经办法是第 07 章的 Alembic。

15. 跑一下本章 demo

完整代码在 demo/demo_models.py，单文件自足：五个模型 + 建表 + 插入 + 回读验证。

uv add sqlalchemy aiosqlite     # 装依赖（greenlet 会随 SQLAlchemy 自动装上）
python demo_models.py

实际运行输出（真实验证过）：

[1] 建表完成: article, article_tag, article_version, channel, tag
[2] 插入完成: article.id=1（自增主键）, channel=Channel(id=1, name='后端进阶'), tags=['SQLAlchemy', 'Python']
[3] 回读验证: 《SQLAlchemy 2.0 建模入门》
    栏目 slug : backend（外键 channel_id=1）
    标签     : ['SQLAlchemy', 'Python']（article_tag 关联表自动落行）
    版本     : [(1, '（草稿）建模入门'), (2, 'SQLAlchemy 2.0 建模入门')]
    创建时间 : 2026-08-27 15:16:52（server_default=func.now() 由数据库填充）

注意 demo 里两个值得回头看一眼的细节：

查询回读用了 selectinload 预加载关系——异步会话里不能懒加载（第 9 节的坑）；
meta 列用了 JSON().with_variant(JSONB(), "postgresql")——SQLite 上跑 demo，PG 上生产，一套代码两用。

16. 小结速查

需求	用法
声明模型基类	`class Base(DeclarativeBase)`，一个项目一个
自动表名	`@declared_attr.directive` 返回 `cls.__name__.lower()`
定义列	`Mapped[类型] = mapped_column(...)`
可空列	注解写 `X \\| None`
Python 端默认值	`default=值` / `default=可调用`
数据库端默认值	`server_default=func.now()`
唯一 / 索引	`unique=True` / `index=True`，组合用 `__table_args__`
外键	`ForeignKey("表.列", ondelete="CASCADE")`
一对多	双向 `relationship(back_populates=...)`
多对多	`Table(...)` 复合主键关联表 + `secondary=`
公共字段	`IDMixin` / `TimestampMixin` 多继承
防循环导入	`if TYPE_CHECKING:` + 字符串注解
PG JSON	`JSONB`，跨库用 `with_variant`
一键建表	`await conn.run_sync(Base.metadata.create_all)`

自测三问：

default 和 server_default 各在什么环节生效？绕过 ORM 手写 INSERT 时，哪个还有效？
多对多的关联表，什么时候用 Table + secondary，什么时候要升级成模型类？
为什么说 create_all 不能替代 Alembic？它对"已存在的表"做什么？

表已经建好，数据怎么进去、怎么查出来、怎么改——下一章把 SQL 表达式的十八般武艺过一遍。

下一章：06 · SQL 表达式与数据操作

1. 心智模型：类是图纸，实例是砖

2. DeclarativeBase：所有模型的"户籍管理处"

3. __tablename__：给表起名字

4. Mapped + mapped_column：类型标注驱动的列定义

5. 主键与自增

6. 可空与默认值：default 和 server_default 是两套机制

7. 唯一与索引：unique、index、__table_args__

8. ForeignKey 与 ondelete：外键及其"应急预案"

9. 一对多：relationship + back_populates

10. 多对多：secondary 关联表与复合主键

11. 模型拆文件时：TYPE_CHECKING 防循环导入

12. Mixin：公共字段的"预制件"

13. PG 特有类型：以 JSONB 为代表

14. create_all：一键建表，但它不是迁移工具

15. 跑一下本章 demo

16. 小结速查

3. `tablename`：给表起名字

4. `Mapped` + `mapped_column`：类型标注驱动的列定义

6. 可空与默认值：`default` 和 `server_default` 是两套机制

7. 唯一与索引：`unique`、`index`、`__table_args__`

8. `ForeignKey` 与 `ondelete`：外键及其"应急预案"

9. 一对多：`relationship` + `back_populates`

10. 多对多：`secondary` 关联表与复合主键

11. 模型拆文件时：`TYPE_CHECKING` 防循环导入

14. `create_all`：一键建表，但它不是迁移工具