14 ORM 建表 —— 把 Python 类变成 MySQL 表，只差一个 create_all

前置阅读: 13-SQLAlchemyORM
关键词: DeclarativeBase, Mapped, mapped_column, ForeignKey, run_sync
难度: ★★★★☆

场景导入：模型声明和 DDL，能不能写在同一处？

第 13 章已经把 Engine、Session 工厂和 DeclarativeBase 准备好了。现在要做的事很具体：声明 Author（作者）和 Book（图书）两个模型，让 Python 类型、列约束和表关系在同一个类里描述清楚，然后在应用启动时通过异步连接完成建表。

SQLAlchemy 2.x 的做法是把这些信息全部收敛到模型类中——Mapped[int] 表达"整数列"，mapped_column(primary_key=True) 补充"这是主键"，ForeignKey("authors.id") 声明外键引用，relationship 提供对象间的导航能力。你不再需要单独写 DDL 脚本——Base.metadata.create_all 会检查所有已注册的模型，生成对应的 CREATE TABLE 语句。本篇只声明 Author 和 Book，Borrow（借阅记录）留到第 20 章引入。

原理解析：从 Python 类到数据库列的映射规则

SQLAlchemy 2.x 的声明式模型由 DeclarativeBase 统一注册。模型继承 Base 后，__tablename__ 指定数据库表名，每个 Mapped[...] 属性对应一列。最常用的映射规则如下：

ForeignKey 和 relationship 是两个容易搞混的概念——前者是数据库层约束，后者是 ORM 层对象导航：

（图注：ForeignKey 约束 Book.author_id 必须引用已存在的 authors.id；双向 relationship 让 author.books 和 book.author 可以在 Python 代码中互相导航。）

Book.author_id 使用 ForeignKey("authors.id", ondelete="CASCADE") 声明数据库层引用完整性：图书必须指向已存在的作者，删除作者时数据库自动级联删除其所有图书。relationship 不创建外键——它只是 ORM 层的导航声明，让 book.author 能点出作者对象、author.books 能列出所有图书。两侧通过 back_populates 互相绑定，设置一侧时 SQLAlchemy 自动同步另一侧的内存关系。

Book.status 使用 Python Enum 约束取值。代码中声明 Enum(BookStatus, name="book_status") 后，MySQL 支持原生枚举时会生成 ENUM('AVAILABLE', 'BORROWED', 'RESERVED')——数据库层强制约束，新增状态值需要迁移 DDL。如果关闭原生枚举，退化为 VARCHAR 加检查约束，变更更灵活但约束力更弱。

建表流程的关键是 run_sync。同步的 Base.metadata.create_all 不能直接接收 AsyncEngine——需要先用 async with engine.begin() as conn 取得异步连接和事务，再执行 await conn.run_sync(Base.metadata.create_all)。run_sync 把同步的元数据操作桥接到异步连接所代理的同步 DBAPI 连接上，实际数据库 I/O 仍经 aiomysql 完成。

（图注：启动流程全程沿异步路径进入连接，仅由 run_sync 适配元数据 DDL 接口，不创建第二套同步引擎。）

代码实现：Author 和 Book，两个模型搞定

先声明作者模型。books 不是表字段，而是 ORM 维护的一对多集合；cascade="all, delete-orphan" 表示删除作者时级联删除所有图书，从 author.books 集合中移除的图书也会被删除：

# app/models/author.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db import Base


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(64), nullable=False)
    bio: Mapped[str | None] = mapped_column(String(512))

    books: Mapped[list["Book"]] = relationship(
        "Book",
        back_populates="author",
        cascade="all, delete-orphan",
    )

图书模型保留 id、title、author_id、isbn、status 五个字段。author_id 的 index=True 支持按作者筛选时走索引，isbn 的 unique=True 阻止重复登记，status 的 default=BookStatus.AVAILABLE 保证新书初始为可借：

# app/models/book.py
import enum

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

from app.db import Base


class BookStatus(str, enum.Enum):
    AVAILABLE = "AVAILABLE"
    BORROWED = "BORROWED"
    RESERVED = "RESERVED"


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(128), nullable=False)
    author_id: Mapped[int] = mapped_column(
        ForeignKey("authors.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    isbn: Mapped[str] = mapped_column(String(13), unique=True, nullable=False)
    status: Mapped[BookStatus] = mapped_column(
        Enum(BookStatus, name="book_status"),
        default=BookStatus.AVAILABLE,
        nullable=False,
    )

    author: Mapped["Author"] = relationship(
        "Author",
        back_populates="books",
    )

模型包集中导入，让 init_db 中的 from app import models 能触发所有表的注册：

# app/models/__init__.py
from app.models.author import Author
from app.models.book import Book

__all__ = ["Author", "Book"]

app/db.py 沿用第 13 章的完整实现，这里不重复列出。建表前必须导入模型包——否则 Base.metadata 为空，create_all 静默返回但一张表也不创建。这是初学者最容易踩的坑。

避坑指南

• ForeignKey 和 relationship 职责不同。 前者生成数据库约束（FOREIGN KEY ... REFERENCES ...），后者提供 Python 侧的对象导航。只写 relationship 不写 ForeignKey，数据库无法阻止悬挂引用；只写 ForeignKey 不写 relationship，Python 代码里无法用 .author 和 .books 导航。

• cascade="all, delete-orphan" 要慎用。 它意味着从父对象集合中移除子对象就会触发数据库删除。如果图书需要保留审计记录，删除作者时不应级联删除图书——改用 ondelete="SET NULL" 或保留软删除行。

• 枚举变更必须迁移。 原生 ENUM 新增或删除值会修改表结构，改 Python 枚举类本身不会自动升级已有列。上线前要验证旧数据、排序语义和回滚路径。

• create_all 不是迁移工具。 它只创建缺失的表，不会修改已存在的列、索引或约束。开发期可以快速建空库，生产环境必须用 Alembic 生成、审查并按版本执行迁移。

• 默认 lazy loading 是 N+1 的祸根。 遍历图书后逐个访问 book.author 会产生额外 SELECT。列表查询应显式使用 selectinload 或 joinedload，异步代码中更不要在不可 await 的位置触发隐式 I/O。

面试 QA

Q1 [源码]: ForeignKey 与 relationship 的区别与协作方式是什么？

ForeignKey 是列级数据库约束，relationship 是 ORM 映射属性。前者保证引用完整性并参与 DDL——ForeignKey("authors.id") 会让数据库拒绝插入指向不存在的作者的图书。后者让业务代码可以沿对象关系导航——book.author.name 能直接拿到作者名字。

源码上，ForeignKey 属于 SQLAlchemy Core 的 schema 定义，relationship 属于 ORM 的映射配置，两者最终由 Mapper 根据外键路径推导连接条件。在异步栈下，relationship 触发的 lazy loading 仍会发出额外 SELECT——所以实际项目应优先显式预加载，而不是依赖隐式查询。

Q2 [项目]: 生产环境如何使用 Alembic 管理异步数据库迁移？

生产环境绝不能依赖启动期 create_all——它不会修改已有表结构。Alembic 是 SQLAlchemy 生态的标准迁移工具，异步配置的关键是 env.py 中用 create_async_engine(DATABASE_URL) 创建引擎，再以 connection.run_sync(...) 包裹 Alembic 的同步迁移入口。

发布流程应把迁移和应用发布解耦：先审查自动生成的 upgrade() 与 downgrade()，再由 CI/CD 流水线执行迁移。数据迁移和 schema 迁移也应分离——先回填新列、再切换代码、最后删除旧列。每个版本都要准备并演练回滚路径。

小结

本章用 Mapped 和 mapped_column 声明了 Author 和 Book，以外键和双向关系表达了一对多模型，并通过 engine.begin() 与 run_sync 在启动期完成异步建表。app/db.py 继续沿用第 13 章的统一基础设施，生产环境的 schema 演进交给 Alembic。

下一篇《15 路由匹配中使用 ORM》将在路由中注入异步 Session，执行第一组真实的数据库查询——GET /books 不再返回空列表，而是从 MySQL 读取 Book 实例。