SqlAlchemy 2.0 中文文档(十八)(2)

简介: SqlAlchemy 2.0 中文文档(十八)

SqlAlchemy 2.0 中文文档(十八)(1)https://developer.aliyun.com/article/1562920


配置映射器级别的“提前加载”行为

在 使用 raiseload 防止延迟列加载 中首次引入的 “raiseload” 行为也可以作为默认的映射器级别行为应用,使用 mapped_column.deferred_raiseload 参数的 mapped_column()。当使用此参数时,受影响的列将在所有情况下在访问时引发,除非在查询时显式地使用 undefer()load_only() 进行“取消延迟”:

>>> class Book(Base):
...     __tablename__ = "book"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     owner_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...     title: Mapped[str]
...     summary: Mapped[str] = mapped_column(Text, deferred=True, deferred_raiseload=True)
...     cover_photo: Mapped[bytes] = mapped_column(
...         LargeBinary, deferred=True, deferred_raiseload=True
...     )
...
...     def __repr__(self) -> str:
...         return f"Book(id={self.id!r}, title={self.title!r})"

使用上述映射,.summary.cover_photo 列默认情况下不可加载:

>>> book = session.scalar(select(Book).where(Book.id == 2))
SELECT  book.id,  book.owner_id,  book.title
FROM  book
WHERE  book.id  =  ?
[...]  (2,)
>>> book.summary
Traceback (most recent call last):
...
sqlalchemy.exc.InvalidRequestError: 'Book.summary' is not available due to raiseload=True

只有在查询时重写它们的行为,通常使用 undefer()undefer_group(),或者更少见的 defer(),属性才能被加载。下面的示例将 undefer('*') 应用于未延迟加载所有属性,并且还利用了填充现有对象来刷新已加载对象的加载器选项:

>>> book = session.scalar(
...     select(Book)
...     .where(Book.id == 2)
...     .options(undefer("*"))
...     .execution_options(populate_existing=True)
... )
SELECT  book.id,  book.owner_id,  book.title,  book.summary,  book.cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,)
>>> book.summary
'another long summary'
```## 将任意 SQL 表达式加载到对象上
如选择 ORM 实体和属性及其他地方所讨论的,可以使用 `select()` 结构在结果集中加载任意 SQL 表达式。比如,如果我们想要发出一个查询,加载 `User` 对象,但也包括每个 `User` 拥有多少书籍的计数,我们可以使用 `func.count(Book.id)` 将“计数”列添加到一个查询中,该查询包括与 `Book` 的 JOIN 以及按所有者 id 进行的 GROUP BY。这将产生 `Row` 对象,每个对象包含两个条目,一个是 `User`,一个是 `func.count(Book.id)`:
```py
>>> 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.book_count 添加到 User 中。这个 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 语句的最外层级。如果在 select() 中使用,而后将其用作子查询或作为复合语句中的元素,如 UNION,它将不起作用。

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

在下面的示例中,使用两个 select() 构造针对带有额外 SQL 表达式标记为 expr 的 ORM 实体 A,并使用 union_all() 组合。然后,在最顶层,从此 UNION 中 SELECT A 实体,使用在 从 UNION 和其他集合操作中选择实体 中描述的查询技术,添加一个选项,使用 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

列加载 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
    布尔值,如果为 True,则表示如果执行加载操作,则应引发异常。
    1.4 版中的新内容。

额外的参数与 column_property() 相同。

另请参阅

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

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,只加载 namefullname 属性:

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 表达式应用于“延迟表达式”属性。

此选项与query_expression() mapper-level 构造一起使用,指示应该是临时 SQL 表达式目标的属性。

例如:

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

版本 1.2 中的新增内容。

参数:

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

另请参阅

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


SqlAlchemy 2.0 中文文档(十八)(3)https://developer.aliyun.com/article/1562922

相关文章
|
5月前
|
SQL 关系型数据库 MySQL
SqlAlchemy 2.0 中文文档(十七)(2)
SqlAlchemy 2.0 中文文档(十七)
46 4
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(十七)(3)
SqlAlchemy 2.0 中文文档(十七)
43 4
|
5月前
|
SQL 关系型数据库 API
SqlAlchemy 2.0 中文文档(十七)(4)
SqlAlchemy 2.0 中文文档(十七)
94 4
|
5月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档(十九)(2)
SqlAlchemy 2.0 中文文档(十九)
44 2
|
5月前
|
SQL 测试技术 Go
SqlAlchemy 2.0 中文文档(十八)(5)
SqlAlchemy 2.0 中文文档(十八)
31 1
|
5月前
|
SQL 测试技术 Go
SqlAlchemy 2.0 中文文档(十八)(3)
SqlAlchemy 2.0 中文文档(十八)
31 1
|
5月前
|
SQL 前端开发 Go
SqlAlchemy 2.0 中文文档(十八)(4)
SqlAlchemy 2.0 中文文档(十八)
32 1
|
5月前
|
SQL 存储 大数据
SqlAlchemy 2.0 中文文档(十八)(1)
SqlAlchemy 2.0 中文文档(十八)
44 1
|
5月前
|
SQL Java Go
SqlAlchemy 2.0 中文文档(十九)(1)
SqlAlchemy 2.0 中文文档(十九)
46 1
|
5月前
|
SQL 关系型数据库 MySQL
SqlAlchemy 2.0 中文文档(十七)(5)
SqlAlchemy 2.0 中文文档(十七)
35 1