SqlAlchemy 2.0 中文文档（十八）（5）-阿里云开发者社区

SqlAlchemy 2.0 中文文档（十八）（4）https://developer.aliyun.com/article/1562923

加载任意 SQL 表达式到对象上

如在选择 ORM 实体和属性和其他地方讨论的，select() 构造可以用于在结果集中加载任意 SQL 表达式。例如，如果我们想要发出一个查询，加载 User 对象，但还包括每个 User 拥有多少书籍的计数，我们可以使用 func.count(Book.id) 来向查询中添加一个“计数”列，该查询包括与 Book 的 JOIN 以及按所有者 id 分组。这将产生包含两个条目的 Row 对象，一个是 User，另一个是 func.count(Book.id)：

>>> from sqlalchemy import func
>>> stmt = select(User, func.count(Book.id)).join_from(User, Book).group_by(Book.owner_id)
>>> for user, book_count in session.execute(stmt):
...     print(f"Username: {user.name}  Number of books: {book_count}")
SELECT  user_account.id,  user_account.name,  user_account.fullname,
count(book.id)  AS  count_1
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
GROUP  BY  book.owner_id
[...]  ()
Username: spongebob  Number of books: 3
Username: sandy  Number of books: 3

在上面的例子中，User 实体和 “书籍数量” SQL 表达式是分开返回的。然而，一个常见的用例是生成一个查询，该查询仅返回 User 对象，例如可以使用 Session.scalars() 进行迭代，其中 func.count(Book.id) SQL 表达式的结果被动态应用于每个 User 实体。最终结果将类似于使用 column_property() 将任意 SQL 表达式映射到类的情况，只不过 SQL 表达式可以在查询时修改。对于这种用例，SQLAlchemy 提供了 with_expression() 加载器选项，当与映射器级别的 query_expression() 指令结合使用时，可能会产生此结果。

要将 with_expression() 应用于查询，映射类必须预先使用 query_expression() 指令配置好一个 ORM 映射属性；此指令将在映射类上生成一个适合接收查询时 SQL 表达式的属性。下面我们向 User 添加一个新属性 User.book_count。这个 ORM 映射属性是只读的，没有默认值；在加载的实例上访问它通常会返回 None：

>>> from sqlalchemy.orm import query_expression
>>> class User(Base):
...     __tablename__ = "user_account"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     fullname: Mapped[Optional[str]]
...     book_count: Mapped[int] = query_expression()
...
...     def __repr__(self) -> str:
...         return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

在我们的映射中配置了 User.book_count 属性后，我们可以使用 with_expression() 加载器选项从 SQL 表达式中填充数据，以便在加载每个 User 对象时应用自定义 SQL 表达式：

>>> from sqlalchemy.orm import with_expression
>>> stmt = (
...     select(User)
...     .join_from(User, Book)
...     .group_by(Book.owner_id)
...     .options(with_expression(User.book_count, func.count(Book.id)))
... )
>>> for user in session.scalars(stmt):
...     print(f"Username: {user.name}  Number of books: {user.book_count}")
SELECT  count(book.id)  AS  count_1,  user_account.id,  user_account.name,
user_account.fullname
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
GROUP  BY  book.owner_id
[...]  ()
Username: spongebob  Number of books: 3
Username: sandy  Number of books: 3

在上面的例子中，我们将我们的 func.count(Book.id) 表达式从 select() 构造函数的 columns 参数中移出，并将其放入 with_expression() 加载器选项中。ORM 然后将其视为一个特殊的列加载选项，该选项动态应用于语句。

query_expression() 映射有以下注意事项：

在一个对象上，如果没有使用 with_expression() 来填充属性，对象实例上的属性将具有值 None，除非在映射上设置了 query_expression.default_expr 参数为默认 SQL 表达式。
with_expression() 值不会填充已加载的对象，除非使用 Populate Existing。如下示例不会起作用，因为 A 对象已经加载：

# load the first A
obj = session.scalars(select(A).order_by(A.id)).first()
# load the same A with an option; expression will **not** be applied
# to the already-loaded object
obj = session.scalars(select(A).options(with_expression(A.expr, some_expr))).first()

要确保属性在现有对象上重新加载，请使用 Populate Existing 执行选项以确保重新填充所有列：

obj = session.scalars(
    select(A)
    .options(with_expression(A.expr, some_expr))
    .execution_options(populate_existing=True)
).first()

当对象过期时，with_expression() SQL 表达式会丢失。一旦对象过期，无论是通过 Session.expire() 还是通过 Session.commit() 的 expire_on_commit 行为，SQL 表达式及其值将不再与属性关联，并且在后续访问时将返回 None。
with_expression() 作为对象加载选项，只对查询的最外层部分生效，并且仅适用于对完整实体进行的查询，而不适用于子查询中的任意列选择或复合语句（如 UNION）的元素。请参阅下一节使用 with_expression() 与 UNIONs、其他子查询中的示例。
映射属性无法应用于查询的其他部分，例如 WHERE 子句、ORDER BY 子句，并利用临时表达式；也就是说，以下方式行不通：

# can't refer to A.expr elsewhere in the query
stmt = (
    select(A)
    .options(with_expression(A.expr, A.x + A.y))
    .filter(A.expr > 5)
    .order_by(A.expr)
)

在上述的 WHERE 子句和 ORDER BY 子句中，A.expr 表达式将会解析为 NULL。要在整个查询中使用该表达式，请将其赋值给一个变量并使用它：

# assign desired expression up front, then refer to that in
# the query
a_expr = A.x + A.y
stmt = (
    select(A)
    .options(with_expression(A.expr, a_expr))
    .filter(a_expr > 5)
    .order_by(a_expr)
)

另请参见

with_expression() 选项是一种特殊选项，用于在查询时动态地将 SQL 表达式应用于映射类。对于在映射器上配置的普通固定 SQL 表达式，请参阅作为映射属性的 SQL 表达式部分。

使用 `with_expression()` 与 UNIONs、其他子查询

with_expression() 构造是一个 ORM 加载器选项，因此只能应用于用于加载特定 ORM 实体的 SELECT 语句的最外层级别。如果在后续用作子查询或复合语句（如 UNION）中使用，它将不起作用。

为了在子查询中使用任意的 SQL 表达式，应该使用正常的 Core 风格添加表达式的方法。要将子查询派生的表达式组装到 ORM 实体的query_expression()属性上，需要在 ORM 对象加载的顶层使用with_expression()，引用子查询中的 SQL 表达式。

在下面的示例中，针对 ORM 实体 A 使用了两个select() 构造，其中包含一个标记为 expr 的额外 SQL 表达式，并使用union_all() 进行组合。然后，在最顶层，从这个 UNION 中选择了 A 实体，使用了在从 UNIONs 和其他集合操作中选择实体中描述的查询技术，添加了一个使用with_expression()的选项，以将这个 SQL 表达式提取到新加载的 A 实例中：

>>> from sqlalchemy import union_all
>>> s1 = (
...     select(User, func.count(Book.id).label("book_count"))
...     .join_from(User, Book)
...     .where(User.name == "spongebob")
... )
>>> s2 = (
...     select(User, func.count(Book.id).label("book_count"))
...     .join_from(User, Book)
...     .where(User.name == "sandy")
... )
>>> union_stmt = union_all(s1, s2)
>>> orm_stmt = (
...     select(User)
...     .from_statement(union_stmt)
...     .options(with_expression(User.book_count, union_stmt.selected_columns.book_count))
... )
>>> for user in session.scalars(orm_stmt):
...     print(f"Username: {user.name}  Number of books: {user.book_count}")
SELECT  user_account.id,  user_account.name,  user_account.fullname,  count(book.id)  AS  book_count
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
WHERE  user_account.name  =  ?
UNION  ALL
SELECT  user_account.id,  user_account.name,  user_account.fullname,  count(book.id)  AS  book_count
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
WHERE  user_account.name  =  ?
[...]  ('spongebob',  'sandy')
Username: spongebob  Number of books: 3
Username: sandy  Number of books: 3
```### 使用 `with_expression()` 与 UNIONs，其他子查询
`with_expression()` 构造是一个 ORM 加载器选项，因此只能应用于要加载特定 ORM 实体的 SELECT 语句的最外层级别。如果在将用作子查询或作为联合等复合语句中的元素的`select()`内部使用，则不会产生任何效果。
为了在子查询中使用任意的 SQL 表达式，应该使用正常的 Core 风格添加表达式的方法。要将子查询派生的表达式组装到 ORM 实体的`query_expression()`属性上，需要在 ORM 对象加载的顶层使用`with_expression()`，引用子查询中的 SQL 表达式。
在下面的示例中，使用两个`select()`构造针对 ORM 实体 `A`，并在`expr`中标记了一个额外的 SQL 表达式，并使用`union_all()`将它们组合起来。然后，在最顶层，使用查询技术描述的 `with_expression()`从这个 UNION 中选择 `A` 实体，以将此 SQL 表达式提取到新加载的 `A` 实例上：
```py
>>> from sqlalchemy import union_all
>>> s1 = (
...     select(User, func.count(Book.id).label("book_count"))
...     .join_from(User, Book)
...     .where(User.name == "spongebob")
... )
>>> s2 = (
...     select(User, func.count(Book.id).label("book_count"))
...     .join_from(User, Book)
...     .where(User.name == "sandy")
... )
>>> union_stmt = union_all(s1, s2)
>>> orm_stmt = (
...     select(User)
...     .from_statement(union_stmt)
...     .options(with_expression(User.book_count, union_stmt.selected_columns.book_count))
... )
>>> for user in session.scalars(orm_stmt):
...     print(f"Username: {user.name}  Number of books: {user.book_count}")
SELECT  user_account.id,  user_account.name,  user_account.fullname,  count(book.id)  AS  book_count
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
WHERE  user_account.name  =  ?
UNION  ALL
SELECT  user_account.id,  user_account.name,  user_account.fullname,  count(book.id)  AS  book_count
FROM  user_account  JOIN  book  ON  user_account.id  =  book.owner_id
WHERE  user_account.name  =  ?
[...]  ('spongebob',  'sandy')
Username: spongebob  Number of books: 3
Username: sandy  Number of books: 3

列加载 API

对象名称	描述
defer(key, *addl_attrs, [raiseload])	指示给定的面向列的属性应该延迟加载，例如在访问之前不加载。
deferred(column, *additional_columns, [group, raiseload, comparator_factory, init, repr, default, default_factory, compare, kw_only, active_history, expire_on_flush, info, doc])	表示默认情况下不加载的基于列的映射属性。
load_only(*attrs, [raiseload])	表示对于特定实体，只应加载给定列表的基于列的属性名称；所有其他属性都将被延迟加载。
query_expression([default_expr], *, [repr, compare, expire_on_flush, info, doc])	表示从查询时 SQL 表达式填充的属性。
undefer(key, *addl_attrs)	指示给定的面向列的属性应该取消延迟加载，例如在整个实体的 SELECT 语句中指定。
undefer_group(name)	表示给定延迟组名内的列应该取消延迟加载。
with_expression(key, expression)	将特定的 SQL 表达式应用于“延迟表达式”属性。

function sqlalchemy.orm.defer(key: Literal['*'] | QueryableAttribute[Any], *addl_attrs: Literal['*'] | QueryableAttribute[Any], raiseload: bool = False) → _AbstractLoad

指示给定的面向列的属性应该延迟加载，例如在访问之前不加载。

此函数是Load接口的一部分，支持方法链和独立操作。

例如：

from sqlalchemy.orm import defer
session.query(MyClass).options(
 defer(MyClass.attribute_one),
 defer(MyClass.attribute_two)
)

要指定相关类上属性的延迟加载，可以逐个指定路径，沿着链指定每个链接的加载样式。要保持链接的加载样式不变，请使用defaultload()：

session.query(MyClass).options(
 defaultload(MyClass.someattr).defer(RelatedClass.some_column)
)

可以使用Load.options()一次捆绑多个与关系相关的延迟选项。

select(MyClass).options(
 defaultload(MyClass.someattr).options(
 defer(RelatedClass.some_column),
 defer(RelatedClass.some_other_column),
 defer(RelatedClass.another_column)
 )
)

参数：

key – 要延迟加载的属性。
raiseload – 当访问延迟属性时引发InvalidRequestError而不是惰性加载值。用于防止发出不必要的 SQL。

版本 1.4 中的新功能。

另请参阅

限制使用列延迟加载 - 在 ORM 查询指南中

load_only()

undefer()

function sqlalchemy.orm.deferred(column: _ORMColumnExprArgument[_T], *additional_columns: _ORMColumnExprArgument[Any], group: str | None = None, raiseload: bool = False, comparator_factory: Type[PropComparator[_T]] | None = None, init: _NoArg | bool = _NoArg.NO_ARG, repr: _NoArg | bool = _NoArg.NO_ARG, default: Any | None = _NoArg.NO_ARG, default_factory: _NoArg | Callable[[], _T] = _NoArg.NO_ARG, compare: _NoArg | bool = _NoArg.NO_ARG, kw_only: _NoArg | bool = _NoArg.NO_ARG, active_history: bool = False, expire_on_flush: bool = True, info: _InfoType | None = None, doc: str | None = None) → MappedSQLExpression[_T]

指示一个基于列的映射属性，默认情况下不会加载，除非访问。

在使用mapped_column()时，通过使用mapped_column.deferred参数，提供了与deferred()构造相同的功能。

参数：

*columns – 要映射的列。通常是单个Column对象，但为了支持在同一属性下映射多个列，也支持集合。
raiseload –
boolean，如果为 True，则表示在执行加载操作时应引发异常。
版本 1.4 中的新功能。

其他参数与column_property()相同。

另请参阅

使用 deferred()为命令式映射器、映射的 SQL 表达式

function sqlalchemy.orm.query_expression(default_expr: _ORMColumnExprArgument[_T] = <sqlalchemy.sql.elements.Null object>, *, repr: Union[_NoArg, bool] = _NoArg.NO_ARG, compare: Union[_NoArg, bool] = _NoArg.NO_ARG, expire_on_flush: bool = True, info: Optional[_InfoType] = None, doc: Optional[str] = None) → MappedSQLExpression[_T]

指示从查询时 SQL 表达式填充的属性。

参数：

default_expr – 可选的 SQL 表达式对象，如果未使用with_expression()分配，则将在所有情况下使用。

版本 1.2 中的新功能。

另请参阅

将任意 SQL 表达式加载到对象上 - 背景和使用示例

function sqlalchemy.orm.load_only(*attrs: Literal['*'] | QueryableAttribute[Any], raiseload: bool = False) → _AbstractLoad

指示对于特定实体，只应加载给定的列名列表的基于列的属性；所有其他属性将被延迟加载。

此函数是Load接口的一部分，支持方法链接和独立操作。

示例 - 给定一个类User，仅加载name和fullname属性：

session.query(User).options(load_only(User.name, User.fullname))

示例 - 给定一个关系User.addresses -> Address，为User.addresses集合指定子查询加载，但在每个Address对象上仅加载email_address属性：

session.query(User).options(
 subqueryload(User.addresses).load_only(Address.email_address)
)

对于具有多个实体的语句，可以使用Load构造函数来特定引用主实体：

stmt = (
 select(User, Address)
 .join(User.addresses)
 .options(
 Load(User).load_only(User.name, User.fullname),
 Load(Address).load_only(Address.email_address),
 )
)

与 populate_existing 执行选项一起使用时，只会刷新列出的属性。

参数：

*attrs – 需要加载的属性，其他所有属性都将被延迟加载。
raiseload –
当访问延迟属性时引发 InvalidRequestError 而不是懒加载值。用于防止不必要的 SQL 发出。
2.0 版本中新增。

参见

限制加载哪些列与列延迟 - 在 ORM 查询指南中

参数：

*attrs – 需要加载的属性，其他所有属性都将被延迟加载。
raiseload –
当访问延迟属性时引发 InvalidRequestError 而不是懒加载值。用于防止不必要的 SQL 发出。
2.0 版本中新增。

function sqlalchemy.orm.undefer(key: Literal['*'] | QueryableAttribute[Any], *addl_attrs: Literal['*'] | QueryableAttribute[Any]) → _AbstractLoad

表示给定的面向列的属性应该取消延迟加载，例如在整个实体的 SELECT 语句中指定。

通常在映射上设置的列作为 deferred() 属性。

此函数是 Load 接口的一部分，支持方法链接和独立操作。

示例：

# undefer two columns
session.query(MyClass).options(
 undefer(MyClass.col1), undefer(MyClass.col2)
)
# undefer all columns specific to a single class using Load + *
session.query(MyClass, MyOtherClass).options(
 Load(MyClass).undefer("*")
)
# undefer a column on a related object
select(MyClass).options(
 defaultload(MyClass.items).undefer(MyClass.text)
)

参数：

key – 需要取消延迟加载的属性。

参见

限制加载哪些列与列延迟 - 在 ORM 查询指南中

defer()

undefer_group()

function sqlalchemy.orm.undefer_group(name: str) → _AbstractLoad

表示给定延迟组名称内的列应取消延迟加载。

正在取消延迟加载的列设置在映射上作为deferred()属性，并包括一个“组”名称。

例如：

session.query(MyClass).options(undefer_group("large_attrs"))

要在相关实体上取消一组属性的延迟加载，可以使用关系加载器选项拼写出路径，例如 defaultload():

select(MyClass).options(
 defaultload("someattr").undefer_group("large_attrs")
)

参见

限制加载哪些列与列延迟 - 在 ORM 查询指南中

defer()

undefer()

function sqlalchemy.orm.with_expression(key: _AttrType, expression: _ColumnExpressionArgument[Any]) → _AbstractLoad

对“延迟表达式”属性应用临时 SQL 表达式。

此选项与指示应该成为临时 SQL 表达式目标的属性的 query_expression() 映射器级构造一起使用。

例如：

stmt = select(SomeClass).options(
 with_expression(SomeClass.x_y_expr, SomeClass.x + SomeClass.y)
)

1.2 版本中新增。

参数：

key – 需要填充的属性
expr – 应用于属性的 SQL 表达式。

参见

将任意的 SQL 表达式加载到对象上 - 背景和使用示例

)
 .options(
 Load(User).load_only(User.name, User.fullname),
 Load(Address).load_only(Address.email_address),
 )
 )

与 populate_existing 执行选项一起使用时，只会刷新列出的属性。
参数：
+   `*attrs` – 需要加载的属性，其他所有属性都将被延迟加载。
+   `raiseload` –
    当访问延迟属性时引发 `InvalidRequestError` 而不是懒加载值。用于防止不必要的 SQL 发出。
    2.0 版本中新增。
参见
限制加载哪些列与列延迟 - 在 ORM 查询指南 中
参数：
+   `*attrs` – 需要加载的属性，其他所有属性都将被延迟加载。
+   `raiseload` –
    当访问延迟属性时引发 `InvalidRequestError` 而不是懒加载值。用于防止不必要的 SQL 发出。
    2.0 版本中新增。
```py
function sqlalchemy.orm.undefer(key: Literal['*'] | QueryableAttribute[Any], *addl_attrs: Literal['*'] | QueryableAttribute[Any]) → _AbstractLoad

表示给定的面向列的属性应该取消延迟加载，例如在整个实体的 SELECT 语句中指定。

通常在映射上设置的列作为 deferred() 属性。

此函数是 Load 接口的一部分，支持方法链接和独立操作。

示例：

# undefer two columns
session.query(MyClass).options(
 undefer(MyClass.col1), undefer(MyClass.col2)
)
# undefer all columns specific to a single class using Load + *
session.query(MyClass, MyOtherClass).options(
 Load(MyClass).undefer("*")
)
# undefer a column on a related object
select(MyClass).options(
 defaultload(MyClass.items).undefer(MyClass.text)
)

参数：

key – 需要取消延迟加载的属性。

参见

限制加载哪些列与列延迟 - 在 ORM 查询指南中

defer()

undefer_group()

function sqlalchemy.orm.undefer_group(name: str) → _AbstractLoad

表示给定延迟组名称内的列应取消延迟加载。

正在取消延迟加载的列设置在映射上作为deferred()属性，并包括一个“组”名称。

例如：

session.query(MyClass).options(undefer_group("large_attrs"))

要在相关实体上取消一组属性的延迟加载，可以使用关系加载器选项拼写出路径，例如 defaultload():

select(MyClass).options(
 defaultload("someattr").undefer_group("large_attrs")
)

参见

限制加载哪些列与列延迟 - 在 ORM 查询指南中

defer()

undefer()

function sqlalchemy.orm.with_expression(key: _AttrType, expression: _ColumnExpressionArgument[Any]) → _AbstractLoad

对“延迟表达式”属性应用临时 SQL 表达式。

此选项与指示应该成为临时 SQL 表达式目标的属性的 query_expression() 映射器级构造一起使用。

例如：

stmt = select(SomeClass).options(
 with_expression(SomeClass.x_y_expr, SomeClass.x + SomeClass.y)
)

1.2 版本中新增。

参数：

key – 需要填充的属性
expr – 应用于属性的 SQL 表达式。

参见

将任意的 SQL 表达式加载到对象上 - 背景和使用示例

SqlAlchemy 2.0 中文文档（十八）（5）

加载任意 SQL 表达式到对象上

使用 `with_expression()` 与 UNIONs、其他子查询

列加载 API

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

SqlAlchemy 2.0 中文文档（十八）（5）

加载任意 SQL 表达式到对象上

使用 with_expression() 与 UNIONs、其他子查询

列加载 API

热门文章

最新文章

相关课程

相关电子书

使用 `with_expression()` 与 UNIONs、其他子查询