表关系与外键

单独一张表能存的信息有限。真实数据往往彼此关联：一个作者写多篇文章，一个标签被多篇文章使用。关系型数据库用「外键」和「关系」表达这些联系，这是它区别于文档库的最大优势。

35-1 外键 ForeignKey 是什么

外键（Foreign Key）就是「指向另一张表主键的字段」。比如文章表想记录「谁写的」，就在 articles 表里放一个 author_id，它的值必须等于 users 表里的某个 id。

from sqlalchemy import ForeignKey, String, Integer, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    # 外键：值必须对应 users.id，建表时数据库会加约束
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

ForeignKey("users.id") 里的写法不是「类名.属性」，而是「表名.字段名」，这是初学者最容易踩的坑。

35-2 一对多：relationship 让访问更自然

光有外键，查询时还得手动 where(author_id == ...)。SQLAlchemy 的 relationship 能把关联「封装」成对象属性，让你像访问普通属性一样拿到关联数据。

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    # 一个用户有多篇文章
    articles: Mapped[list["Article"]] = relationship(back_populates="author")

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    # 一篇文章属于一个用户
    author: Mapped["User"] = relationship(back_populates="articles")

back_populates 把两边连起来，形成双向关系。现在可以这样用：

user = db.get(User, 1)
print(user.articles)        # 直接拿到该用户的所有文章（列表）

article = db.get(Article, 5)
print(article.author.name)  # 直接拿到作者名字

35-3 一对多怎么增删

有了关系，添加文章时既可以填 author_id，也可以直接挂到 user.articles 上，两者等价：

# 方式一：填外键值
a = Article(title="你好", author_id=1)

# 方式二：挂到关系属性（更直观）
user = db.get(User, 1)
user.articles.append(Article(title="你好"))
db.add(user)
db.commit()

查询某用户的文章：

from sqlalchemy import select
stmt = select(Article).where(Article.author_id == 1)
articles = db.execute(stmt).scalars().all()

35-4 多对多：需要一张关联表

「一对多」解决「一个用户多篇文章」。但「多对多」更复杂：一个文章有多个标签，一个标签也被多篇文章使用。这种情况谁也放不下对方的外键，必须单独建一张「中间表」来存两两配对。

from sqlalchemy import Table, Column, Integer

# 关联表：没有业务逻辑，只记录 article_id 和 tag_id 的配对
article_tag = Table(
    "article_tag",
    Base.metadata,
    Column("article_id", ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    # secondary 指向关联表，形成多对多
    articles: Mapped[list["Article"]] = relationship(
        secondary=article_tag, back_populates="tags"
    )

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    tags: Mapped[list["Tag"]] = relationship(
        secondary=article_tag, back_populates="articles"
    )

secondary=article_tag 告诉 SQLAlchemy：这两张表通过中间表 article_tag 相连。关联表本身不用写模型类，用 Table 声明即可（两个外键一起做联合主键）。

使用起来和一对多一样直观：

article = db.get(Article, 1)
print(article.tags)         # 这篇文章的所有标签

tag = db.get(Tag, 2)
print(tag.articles)         # 带这个标签的所有文章

35-5 级联 cascade：父删子跟着删

有时希望「删掉用户时，他写的文章也一起删」，而不是因为外键约束报错。这靠 cascade 配置：

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    articles: Mapped[list["Article"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",   # 删用户时级联删文章
    )

cascade="all, delete-orphan" 表示：对用户做的所有操作（增删改）都传给文章；一旦文章失去了所属用户（孤儿），也一并删除。

user = db.get(User, 1)
db.delete(user)
db.commit()       # 该用户的所有文章被自动删除

35-6 回写 backref 与双向同步

前面用 back_populates 是显式双向写法。SQLAlchemy 也提供 backref 简写：在一侧声明，另一侧自动生成反向属性。

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    # 用 backref 自动给 User 加上 .articles 属性
    author: Mapped["User"] = relationship("User", backref="articles")

两种写法效果相同。back_populates 更直观、不易出错，推荐新手用；backref 更省代码。本书统一用 back_populates。

35-7 一个完整小例子

把用户、文章、标签串起来，建表后做一次关联写入。注意这个例子里的 Article 同时拥有一对多的 author 和多对多的 tags 两个关系，实际运行时要把前面几节的定义合并成下面这样（User 同时声明 articles 和 tags）：

from sqlalchemy import Column, ForeignKey, Integer, String, Table, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

article_tags = Table(
    "article_tags", Base.metadata,
    Column("article_id", ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    articles: Mapped[list["Article"]] = relationship(back_populates="author")

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="articles")
    tags: Mapped[list["Tag"]] = relationship(secondary=article_tags, back_populates="articles")

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    articles: Mapped[list["Article"]] = relationship(secondary=article_tags, back_populates="tags")

engine = create_engine("sqlite:///./test.db")

然后就能建表并关联写入：

from sqlalchemy import select
from sqlalchemy.orm import Session

Base.metadata.create_all(engine)

with Session(engine) as session:
    user = User(name="小明")
    tag_py = Tag(name="Python")
    article = Article(title="SQLAlchemy 入门", author=user, tags=[tag_py])
    session.add(article)
    session.commit()

    # 反过来查
    a = session.execute(select(Article)).scalars().first()
    print(a.author.name)     # 小明
    print(a.tags[0].name)    # Python

35-8 加载方式：延迟加载与预先加载

默认情况下，关系属性是「延迟加载」：你访问 user.articles 那一刻，SQLAlchemy 才去查文章表。这在单条数据时不明显，但一旦循环打印多个用户各自的文章，就会触发「1 次查用户 + N 次查文章」的 N+1 问题，查询次数随数据量暴涨。

想一次取回关联数据，用 selectinload 预先加载：

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.articles))
users = db.execute(stmt).scalars().all()
# 此时 user.articles 已经填好，不会再触发额外查询

selectinload 会单独发一条按 id 批量取文章的 SQL，把 N+1 压成 2 次查询。关系多、又常一起用的场景，记得加上它。另一种常用选项是 joinedload，它通过 SQL 的 JOIN 把关联数据一次取回，适合一对一小关系。

在 FastAPI 里还有一个和加载方式紧密相关的坑。如果响应模型里包含了关联字段，而你在会话关闭之后才让 Pydantic 去读取它，延迟加载就会因为会话已失效而报错。解决办法有两个：一是像上面那样用 selectinload 在查询阶段就把数据取全，二是把 expire_on_commit=False 加到会话工厂上，让对象在提交后仍保留已加载的值。涉及关联数据的接口，优先选第一种，既避免报错，又能顺手消除 N+1 查询。

这一章把表与表的纽带讲清楚了。下一章我们回到「性能」主题：用异步数据库，让接口在高并发下不被数据库拖慢。