SQLAlchemy 1.4 / 2.0 教程

本页是 SQLAlchemy 统一教程 的一部分。

上一篇: 使用 ORM 操作数据 | 下一篇: 进一步阅读

使用 ORM 关联对象

在本节中，我们将介绍另一个重要的 ORM 概念，即 ORM 如何与引用其他对象的映射类交互。在 声明映射类 章节中，映射类示例使用了名为 relationship() 的构造。此构造定义了两个不同映射类之间或从映射类到自身的链接，后者称为自引用关系。

为了描述 relationship() 的基本思想，首先我们将简要回顾映射，省略 mapped_column() 映射和其他指令

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = "user_account"

    # ... mapped_column() mappings

    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    # ... mapped_column() mappings

    user: Mapped["User"] = relationship(back_populates="addresses")

在上面，User 类现在有一个属性 User.addresses，而 Address 类有一个属性 Address.user。relationship() 构造与 Mapped 构造（用于指示类型行为）结合使用，将用于检查映射到 User 和 Address 类的 Table 对象之间的表关系。由于表示 address 表的 Table 对象具有引用 user_account 表的 ForeignKeyConstraint，因此 relationship() 可以明确地确定从 User 类到 Address 类存在 一对多 关系，沿着 User.addresses 关系；user_account 表中的特定行可以被 address 表中的多行引用。

所有一对多关系自然地对应于另一个方向的 多对一 关系，在本例中由 Address.user 指示。relationship.back_populates 参数，如上所示，在引用另一个名称的两个 relationship() 对象上配置，它建立这两个 relationship() 构造应被视为彼此互补；我们将在下一节中看到这是如何实现的。

持久化和加载关系

我们可以首先说明 relationship() 对对象实例的作用。如果我们创建一个新的 User 对象，我们可以注意到当我们访问 .addresses 元素时，会得到一个 Python 列表

>>> u1 = User(name="pkrabs", fullname="Pearl Krabs")
>>> u1.addresses
[]

此对象是 Python list 的 SQLAlchemy 特定版本，它能够跟踪和响应对其所做的更改。当我们访问属性时，集合也会自动出现，即使我们从未将其分配给对象。这类似于在 使用 ORM 单元工作模式插入行 中注意到的行为，其中观察到我们没有显式赋值的基于列的属性也自动显示为 None，而不是像 Python 的通常行为那样引发 AttributeError。

由于 u1 对象仍然是 瞬态 的，并且我们从 u1.addresses 获取的 list 尚未被修改（即追加或扩展），因此它实际上尚未与对象关联，但是当我们对其进行更改时，它将成为 User 对象状态的一部分。

该集合特定于 Address 类，这是唯一可以持久化在其中的 Python 对象类型。使用 list.append() 方法，我们可以添加一个 Address 对象

>>> a1 = Address(email_address="pearl.krabs@gmail.com")
>>> u1.addresses.append(a1)

此时，u1.addresses 集合如预期的那样包含新的 Address 对象

>>> u1.addresses
[Address(id=None, email_address='pearl.krabs@gmail.com')]

当我们关联 Address 对象与 u1 实例的 User.addresses 集合时，还发生了另一个行为，即 User.addresses 关系与 Address.user 关系同步，这样我们不仅可以从 User 对象导航到 Address 对象，还可以从 Address 对象导航回“父” User 对象

>>> a1.user
User(id=None, name='pkrabs', fullname='Pearl Krabs')

这种同步的发生是由于我们在两个 relationship() 对象之间使用了 relationship.back_populates 参数。此参数命名了另一个 relationship()，对于该关系，应发生互补的属性赋值/列表修改。它在另一个方向上也同样有效，也就是说，如果我们创建另一个 Address 对象并分配给其 Address.user 属性，则该 Address 将成为该 User 对象上 User.addresses 集合的一部分

>>> a2 = Address(email_address="pearl@aol.com", user=u1)
>>> u1.addresses
[Address(id=None, email_address='pearl.krabs@gmail.com'), Address(id=None, email_address='pearl@aol.com')]

我们实际上使用了 user 参数作为 Address 构造函数中的关键字参数，这与在 Address 类上声明的任何其他映射属性一样被接受。它等效于事后赋值 Address.user 属性

# equivalent effect as a2 = Address(user=u1)
>>> a2.user = u1

级联对象到 Session 中

现在我们有一个 User 和两个 Address 对象，它们在内存中以双向结构关联，但正如之前在 使用 ORM 单元工作模式插入行 中提到的，这些对象在与 Session 对象关联之前，都处于 瞬态 状态。

我们使用仍在进行的 Session，并注意到当我们对主导 User 对象应用 Session.add() 方法时，相关的 Address 对象也会添加到同一个 Session 中

>>> session.add(u1)
>>> u1 in session
True
>>> a1 in session
True
>>> a2 in session
True

上述行为，其中 Session 接收到 User 对象，并沿着 User.addresses 关系找到相关的 Address 对象，被称为保存-更新级联，并在 ORM 参考文档 级联 中详细讨论。

这三个对象现在处于 挂起 状态；这意味着它们已准备好成为 INSERT 操作的主题，但尚未进行；所有三个对象都尚未分配主键，此外，a1 和 a2 对象都具有一个名为 user_id 的属性，该属性引用具有引用 user_account.id 列的 ForeignKeyConstraint 的 Column；这些属性也都是 None，因为对象尚未与实际数据库行关联

>>> print(u1.id)
None
>>> print(a1.user_id)
None

正是在这个阶段，我们可以看到单元工作过程提供的巨大实用性；回想一下 INSERT 通常会自动生成“values”子句 章节，为了自动将 address.user_id 列与 user_account 行的列相关联，使用了一些详细的语法将行插入到 user_account 和 address 表中。此外，我们必须先为 user_account 行发出 INSERT，然后再为 address 行发出，因为 address 中的行依赖于其在 user_account 中的父行，以便在其 user_id 列中获得值。

当使用 Session 时，所有这些繁琐的工作都为我们处理了，即使是最顽固的 SQL 纯粹主义者也可以从 INSERT、UPDATE 和 DELETE 语句的自动化中受益。当我们 Session.commit() 事务时，所有步骤都以正确的顺序调用，此外，新生成的 user_account 行的主键会适当地应用于 address.user_id 列

>>> session.commit()
INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] ('pkrabs', 'Pearl Krabs')
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[... (insertmanyvalues) 1/2 (ordered; batch not supported)] ('pearl.krabs@gmail.com', 6)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[insertmanyvalues 2/2 (ordered; batch not supported)] ('pearl@aol.com', 6)
COMMIT

加载关系

在上一步中，我们调用了 Session.commit()，它为事务发出了 COMMIT，然后根据 Session.commit.expire_on_commit 使所有对象过期，以便它们在下一个事务中刷新。

当我们接下来访问这些对象上的属性时，我们将看到为行的主属性发出的 SELECT，例如当我们查看 u1 对象新生成的主键时

>>> u1.id
BEGIN (implicit)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name,
user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (6,)
6

u1 User 对象现在有一个持久化的集合 User.addresses，我们也可以访问它。由于此集合由来自 address 表的另一组行组成，因此当我们也访问此集合时，我们再次看到发出了 延迟加载 以检索对象

>>> u1.addresses
SELECT address.id AS address_id, address.email_address AS address_email_address,
address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (6,)
[Address(id=4, email_address='pearl.krabs@gmail.com'), Address(id=5, email_address='pearl@aol.com')]

SQLAlchemy ORM 中的集合和相关属性在内存中是持久的；一旦集合或属性被填充，就不会再发出 SQL，直到该集合或属性 过期。我们可以再次访问 u1.addresses，以及添加或删除项目，这不会产生任何新的 SQL 调用

>>> u1.addresses
[Address(id=4, email_address='pearl.krabs@gmail.com'), Address(id=5, email_address='pearl@aol.com')]

虽然延迟加载发出的加载如果我们不采取明确的步骤来优化它，可能会很快变得昂贵，但延迟加载的网络至少得到了相当好的优化，不会执行冗余工作；由于 u1.addresses 集合被刷新，根据 标识映射，这些实际上与我们已经处理过的 a1 和 a2 对象是相同的 Address 实例，因此我们已完成加载此特定对象图中的所有属性

>>> a1
Address(id=4, email_address='pearl.krabs@gmail.com')
>>> a2
Address(id=5, email_address='pearl@aol.com')

关系如何加载或不加载的问题本身就是一个完整的主题。有关这些概念的更多介绍稍后在本节的 加载器策略 中介绍。

在查询中使用关系

上一节介绍了在使用映射类的实例时 relationship() 构造的行为，上面是 User 和 Address 类的 u1、a1 和 a2 实例。在本节中，我们将介绍 relationship() 应用于映射类的类级别行为时的行为，它在几个方面有助于自动化 SQL 查询的构造。

使用关系进行 Join

显式 FROM 子句和 JOIN 和 设置 ON 子句 章节介绍了使用 Select.join() 和 Select.join_from() 方法来组合 SQL JOIN 子句。为了描述如何在表之间进行 Join，这些方法要么基于表元数据结构中链接两个表的单个明确的 ForeignKeyConstraint 对象的存在推断 ON 子句，否则我们可以提供一个显式的 SQL 表达式构造来指示特定的 ON 子句。

当使用 ORM 实体时，可以使用另一种机制来帮助我们设置 Join 的 ON 子句，即使用我们在用户映射中设置的 relationship() 对象，如 声明映射类 中演示的那样。对应于 relationship() 的类绑定属性可以作为 Select.join() 的单个参数传递，其中它用于同时指示 Join 的右侧和 ON 子句

>>> print(select(Address.email_address).select_from(User).join(User.addresses))
SELECT address.email_address
FROM user_account JOIN address ON user_account.id = address.user_id

如果我们不指定 ON 子句，则 Select.join() 或 Select.join_from() 不会使用映射上的 ORM relationship() 来推断 ON 子句。这意味着，如果我们从 User Join 到 Address 而没有 ON 子句，它可以工作是因为两个映射的 Table 对象之间的 ForeignKeyConstraint，而不是因为 User 和 Address 类上的 relationship() 对象

>>> print(select(Address.email_address).join_from(User, Address))
SELECT address.email_address
FROM user_account JOIN address ON user_account.id = address.user_id

有关如何将 Select.join() 和 Select.join_from() 与 relationship() 构造一起使用的更多示例，请参见 ORM 查询指南 中的 Joins 章节。

另请参阅

Joins，在 ORM 查询指南 中

关系 WHERE 操作符

与 relationship() 一起使用的一些其他类型的 SQL 生成助手通常在构建语句的 WHERE 子句时很有用。请参阅 ORM 查询指南 中的 关系 WHERE 操作符 章节。

另请参阅

关系 WHERE 操作符，在 ORM 查询指南 中

加载器策略

在 加载关系 章节中，我们介绍了这个概念，即当我们使用映射对象的实例时，访问使用 relationship() 映射的属性，在默认情况下，当集合未填充时，将发出 延迟加载，以便加载应存在于此集合中的对象。

延迟加载是最著名的 ORM 模式之一，也是最具争议的模式之一。当内存中的几十个 ORM 对象各自引用少量未加载的属性时，对这些对象进行常规操作可能会引发许多额外的查询，这些查询可能会累积起来（也称为 N+1 问题），更糟糕的是，它们是隐式发出的。这些隐式查询可能不会被注意到，当它们在没有数据库事务可用后或使用替代并发模式（如 asyncio）时尝试时，可能会导致错误，实际上，它们根本无法工作。

与此同时，当延迟加载与正在使用的并发方法兼容且没有引起其他问题时，它是一种非常流行且有用的模式。出于这些原因，SQLAlchemy 的 ORM 非常重视能够控制和优化此加载行为。

最重要的是，有效使用 ORM 延迟加载的第一步是测试应用程序，打开 SQL 回显，并观察发出的 SQL 语句。如果出现大量看似可以更有效地合并为一个的冗余 SELECT 语句，如果对于已经从其 detached 的 Session 中 detached 的对象不适当地发生加载，那么就应该考虑使用加载器策略。

加载器策略表示为可以与 SELECT 语句关联的对象，使用 Select.options() 方法，例如：

for user_obj in session.execute(
    select(User).options(selectinload(User.addresses))
).scalars():
    user_obj.addresses  # access addresses collection already loaded

它们也可以配置为 relationship() 的默认值，使用 relationship.lazy 选项，例如：

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = "user_account"

    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", lazy="selectin"
    )

每个加载器策略对象都会向语句添加某种信息，Session 稍后在决定应如何加载各种属性和/或在访问它们时如何表现时将使用该信息。

以下章节将介绍一些最常用的加载器策略。

另请参阅

关系加载技术 中的两个章节

• 在映射时配置加载器策略 - 有关在 relationship() 上配置策略的详细信息

• 使用加载器选项进行关系加载 - 有关使用查询时加载器策略的详细信息

Selectin Load

现代 SQLAlchemy 中最有用的加载器是 selectinload() 加载器选项。此选项解决了最常见的 “N+1” 问题形式，即一组对象引用相关集合的问题。selectinload() 将确保预先加载完整系列对象的特定集合，使用单个查询。它通过使用 SELECT 形式来实现这一点，在大多数情况下，SELECT 形式可以针对相关表单独发出，而无需引入 JOIN 或子查询，并且仅查询那些集合尚未加载的父对象。下面我们通过加载所有 User 对象及其所有相关的 Address 对象来说明 selectinload()；虽然我们只调用一次 Session.execute()，给定一个 select() 构造，当访问数据库时，实际上会发出两个 SELECT 语句，第二个语句是获取相关的 Address 对象

>>> from sqlalchemy.orm import selectinload
>>> stmt = select(User).options(selectinload(User.addresses)).order_by(User.id)
>>> for row in session.execute(stmt):
...     print(
...         f"{row.User.name}  ({', '.join(a.email_address for a in row.User.addresses)})"
...     )
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account ORDER BY user_account.id
[...] ()
SELECT address.user_id AS address_user_id, address.id AS address_id,
address.email_address AS address_email_address
FROM address
WHERE address.user_id IN (?, ?, ?, ?, ?, ?)
[...] (1, 2, 3, 4, 5, 6)
spongebob  (spongebob@sqlalchemy.org)
sandy  (sandy@sqlalchemy.org, sandy@squirrelpower.org)
patrick  ()
squidward  ()
ehkrabs  ()
pkrabs  (pearl.krabs@gmail.com, pearl@aol.com)

另请参阅

Select IN 加载 - 在 关系加载技术 中

Joined Load

joinedload() 预先加载策略是 SQLAlchemy 中最旧的预先加载器，它使用 JOIN（可能是外连接或内连接，具体取决于选项）增强传递到数据库的 SELECT 语句，然后可以加载相关对象。

joinedload() 策略最适合加载相关的多对一对象，因为这只需要将额外的列添加到无论如何都会获取的主实体行中。为了更高的效率，它还接受一个选项 joinedload.innerjoin，以便可以使用内连接而不是外连接，例如在以下情况中，我们知道所有 Address 对象都关联到一个 User

>>> from sqlalchemy.orm import joinedload
>>> stmt = (
...     select(Address)
...     .options(joinedload(Address.user, innerjoin=True))
...     .order_by(Address.id)
... )
>>> for row in session.execute(stmt):
...     print(f"{row.Address.email_address} {row.Address.user.name}")
SELECT address.id, address.email_address, address.user_id, user_account_1.id AS id_1,
user_account_1.name, user_account_1.fullname
FROM address
JOIN user_account AS user_account_1 ON user_account_1.id = address.user_id
ORDER BY address.id
[...] ()
spongebob@sqlalchemy.org spongebob
sandy@sqlalchemy.org sandy
sandy@squirrelpower.org sandy
pearl.krabs@gmail.com pkrabs
pearl@aol.com pkrabs

joinedload() 也适用于集合，即一对多关系，但是它具有将主行按每个相关项递归地倍增的效果，从而使结果集发送的数据量按嵌套集合和/或较大集合的数量级增长，因此应根据具体情况评估其使用与诸如 selectinload() 之类的其他选项的优劣。

重要的是要注意，外层 Select 语句的 WHERE 和 ORDER BY 条件不针对 joinedload() 呈现的表。在上面，可以在 SQL 中看到，匿名别名应用于 user_account 表，因此在查询中无法直接寻址。此概念在 预先加载连接的禅宗 章节中进行了更详细的讨论。

提示

重要的是要注意，多对一的预先加载通常不是必需的，因为 “N+1” 问题在常见情况下不太普遍。当许多对象都引用同一个相关对象时，例如许多 Address 对象都引用同一个 User 时，对于该 User 对象，SQL 将只发出一次，使用正常的延迟加载。延迟加载例程将在当前的 Session 中按主键查找相关对象，并在可能的情况下不发出任何 SQL。

另请参阅

连接预先加载 - 在 关系加载技术 中

显式 Join + 预先加载

如果我们使用诸如 Select.join() 之类的方法来呈现 JOIN，从而加载 Address 行，同时连接到 user_account 表，我们也可以利用该 JOIN 来预先加载每个返回的 Address 对象上的 Address.user 属性的内容。这本质上是我们正在使用 “连接预先加载”，但自己呈现 JOIN。这种常见的用例可以通过使用 contains_eager() 选项来实现。此选项与 joinedload() 非常相似，不同之处在于它假定我们已经自己设置了 JOIN，而它只是指示 COLUMNS 子句中的其他列应加载到每个返回对象的相关属性中，例如

>>> from sqlalchemy.orm import contains_eager
>>> stmt = (
...     select(Address)
...     .join(Address.user)
...     .where(User.name == "pkrabs")
...     .options(contains_eager(Address.user))
...     .order_by(Address.id)
... )
>>> for row in session.execute(stmt):
...     print(f"{row.Address.email_address} {row.Address.user.name}")
SELECT user_account.id, user_account.name, user_account.fullname,
address.id AS id_1, address.email_address, address.user_id
FROM address JOIN user_account ON user_account.id = address.user_id
WHERE user_account.name = ? ORDER BY address.id
[...] ('pkrabs',)
pearl.krabs@gmail.com pkrabs
pearl@aol.com pkrabs

在上面，我们既在 user_account.name 上过滤了行，又将 user_account 中的行加载到返回行的 Address.user 属性中。如果我们单独应用 joinedload()，我们将得到一个不必要地连接两次的 SQL 查询

>>> stmt = (
...     select(Address)
...     .join(Address.user)
...     .where(User.name == "pkrabs")
...     .options(joinedload(Address.user))
...     .order_by(Address.id)
... )
>>> print(stmt)  # SELECT has a JOIN and LEFT OUTER JOIN unnecessarily
SELECT address.id, address.email_address, address.user_id,
user_account_1.id AS id_1, user_account_1.name, user_account_1.fullname
FROM address JOIN user_account ON user_account.id = address.user_id
LEFT OUTER JOIN user_account AS user_account_1 ON user_account_1.id = address.user_id
WHERE user_account.name = :name_1 ORDER BY address.id

另请参阅

关系加载技术 中的两个章节

• 预先加载连接的禅宗 - 详细描述了上述问题

• 将显式连接/语句路由到预先加载的集合中 - 使用 contains_eager()

Raiseload

另一个值得一提的加载器策略是 raiseload()。此选项用于完全阻止应用程序出现 N+1 问题，方法是使通常是延迟加载的操作改为引发错误。它有两个变体，通过 raiseload.sql_only 选项控制，以阻止需要 SQL 的延迟加载，或者阻止所有 “加载” 操作，包括那些只需要查阅当前 Session 的操作。

使用 raiseload() 的一种方法是在 relationship() 本身配置它，方法是将 relationship.lazy 设置为值 "raise_on_sql"，这样对于特定的映射，某个关系将永远不会尝试发出 SQL

>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import relationship


>>> class User(Base):
...     __tablename__ = "user_account"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     addresses: Mapped[List["Address"]] = relationship(
...         back_populates="user", lazy="raise_on_sql"
...     )


>>> class Address(Base):
...     __tablename__ = "address"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...     user: Mapped["User"] = relationship(back_populates="addresses", lazy="raise_on_sql")

使用这样的映射，应用程序被阻止进行延迟加载，表明特定的查询需要指定加载器策略

>>> u1 = session.execute(select(User)).scalars().first()
SELECT user_account.id FROM user_account
[...] ()
>>> u1.addresses
Traceback (most recent call last):
...
sqlalchemy.exc.InvalidRequestError: 'User.addresses' is not available due to lazy='raise_on_sql'

该异常将指示应预先加载此集合

>>> u1 = (
...     session.execute(select(User).options(selectinload(User.addresses)))
...     .scalars()
...     .first()
... )
SELECT user_account.id
FROM user_account
[...] ()
SELECT address.user_id AS address_user_id, address.id AS address_id
FROM address
WHERE address.user_id IN (?, ?, ?, ?, ?, ?)
[...] (1, 2, 3, 4, 5, 6)

lazy="raise_on_sql" 选项试图智能地处理多对一关系；在上面，如果 Address.user 对象的 Address.user 属性未加载，但该 User 对象在同一 Session 中本地存在，则 “raiseload” 策略不会引发错误。

另请参阅

使用 raiseload 防止不需要的延迟加载 - 在 关系加载技术 中

SQLAlchemy 1.4 / 2.0 教程

下一教程章节： 进一步阅读