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

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

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


配置映射上的列延迟

defer() 的功能作为映射列的默认行为可用,适用于不应在每次查询时无条件加载的列。要配置,请使用 mapped_column.deferred 参数。下面的示例说明了对 Book 应用默认列延迟加载的映射:

>>> 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)
...     cover_photo: Mapped[bytes] = mapped_column(LargeBinary, deferred=True)
...
...     def __repr__(self) -> str:
...         return f"Book(id={self.id!r}, title={self.title!r})"

使用上述映射,对 Book 的查询将自动不包括 summarycover_photo 列:

>>> book = session.scalar(select(Book).where(Book.id == 2))
SELECT  book.id,  book.owner_id,  book.title
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

与所有延迟加载属性一样,当首次访问加载的对象上的延迟加载属性时,默认行为是它们将 延迟加载 它们的值:

>>> img_data = book.cover_photo
SELECT  book.cover_photo  AS  book_cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

defer()load_only() 加载器选项一样,映射器级别的延迟还包括一个选项,即当语句中没有其他选项时,可以发生 raiseload 行为,而不是延迟加载。这允许某些列不会默认加载,并且也永远不会在语句中使用显式指令时延迟加载。请参阅 配置映射器级别的raiseload行为 部分,了解如何配置和使用此行为的背景信息。

使用 deferred() 来命令式映射,映射 SQL 表达式

deferred() 函数是早期的、更通用的“延迟列”映射指令,在引入 SQLAlchemy 的 mapped_column() 构造之前就存在。

deferred() 在配置 ORM 映射器时使用,接受任意的 SQL 表达式或 Column 对象。因此,它适用于非声明性 命令式映射,将其传递给 map_imperatively.properties 字典:

from sqlalchemy import Blob
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import Text
from sqlalchemy.orm import registry
mapper_registry = registry()
book_table = Table(
    "book",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True),
    Column("title", String(50)),
    Column("summary", Text),
    Column("cover_image", Blob),
)
class Book:
    pass
mapper_registry.map_imperatively(
    Book,
    book_table,
    properties={
        "summary": deferred(book_table.c.summary),
        "cover_image": deferred(book_table.c.cover_image),
    },
)

当映射的 SQL 表达式应该在延迟加载时,可以使用 deferred() 代替 column_property()

from sqlalchemy.orm import deferred
class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    firstname: Mapped[str] = mapped_column()
    lastname: Mapped[str] = mapped_column()
    fullname: Mapped[str] = deferred(firstname + " " + lastname)

另请参阅

使用 column_property - 在 SQL 表达式作为映射属性 部分

对声明性表列应用加载、持久性和映射选项 - 在使用声明性进行表配置章节中

使用undefer()来“急切地”加载延迟列

对于默认配置为延迟的映射上的列,undefer()选项将导致任何通常延迟的列都会在前端加载,也就是说,与映射的所有其他列一起加载。例如,我们可以对前面映射中标记为延迟的Book.summary列应用undefer()

>>> from sqlalchemy.orm import undefer
>>> book = session.scalar(select(Book).where(Book.id == 2).options(undefer(Book.summary)))
SELECT  book.id,  book.owner_id,  book.title,  book.summary
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

Book.summary列现在已经被急切加载,并且可以在不发出额外 SQL 的情况下访问:

>>> print(book.summary)
another long summary

按组加载延迟列

通常,当一个列使用mapped_column(deferred=True)进行映射时,当在对象上访问延迟属性时,SQL 将被发出以仅加载该特定列,而不加载其他列,即使映射还有其他被标记为延迟的列也是如此。在延迟属性是应该一次性加载一组属性的情况下,而不是针对每个属性单独发出 SQL 时,可以使用mapped_column.deferred_group参数,它接受一个任意字符串,用于定义要取消延迟的列的通用组:

>>> 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_group="book_attrs"
...     )
...     cover_photo: Mapped[bytes] = mapped_column(
...         LargeBinary, deferred=True, deferred_group="book_attrs"
...     )
...
...     def __repr__(self) -> str:
...         return f"Book(id={self.id!r}, title={self.title!r})"

使用上述映射,访问summarycover_photo将同时使用一个 SELECT 语句加载两个列:

>>> book = session.scalar(select(Book).where(Book.id == 2))
SELECT  book.id,  book.owner_id,  book.title
FROM  book
WHERE  book.id  =  ?
[...]  (2,)
>>> img_data, summary = book.cover_photo, book.summary
SELECT  book.summary  AS  book_summary,  book.cover_photo  AS  book_cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

使用undefer_group()按组取消延迟

如果延迟列配置为使用前一节中引入的mapped_column.deferred_group,则可以使用undefer_group()选项来急切加载整个组,传递要急切加载的组的字符串名称:

>>> from sqlalchemy.orm import undefer_group
>>> book = session.scalar(
...     select(Book).where(Book.id == 2).options(undefer_group("book_attrs"))
... )
SELECT  book.id,  book.owner_id,  book.title,  book.summary,  book.cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

summarycover_photo都可以在不进行额外加载的情况下使用:

>>> img_data, summary = book.cover_photo, book.summary

通配符上的取消延迟

大多数 ORM 加载器选项都接受通配符表达式,用"*"表示,表示该选项应用于所有相关属性。如果一个映射具有一系列延迟列,那么所有这些列都可以一次性进行取消延迟,而不需要使用组名,只需指定通配符即可:

>>> book = session.scalar(select(Book).where(Book.id == 3).options(undefer("*")))
SELECT  book.id,  book.owner_id,  book.title,  book.summary,  book.cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (3,) 

配置映射级别的“raiseload”行为

首次引入的“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'
```### 使用 `deferred()` 进行命令式映射,映射的 SQL 表达式
`deferred()` 函数是早期的、更通用的“延迟列”映射指令,它在引入 `mapped_column()` 构造之前就存在于 SQLAlchemy 中。
在配置 ORM 映射器时使用 `deferred()`,它接受任意的 SQL 表达式或 `Column` 对象。因此,它适用于非声明式的命令式映射,可以将其传递给 `map_imperatively.properties` 字典:
```py
from sqlalchemy import Blob
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import Text
from sqlalchemy.orm import registry
mapper_registry = registry()
book_table = Table(
    "book",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True),
    Column("title", String(50)),
    Column("summary", Text),
    Column("cover_image", Blob),
)
class Book:
    pass
mapper_registry.map_imperatively(
    Book,
    book_table,
    properties={
        "summary": deferred(book_table.c.summary),
        "cover_image": deferred(book_table.c.cover_image),
    },
)

当映射的 SQL 表达式应该以延迟方式加载时,也可以使用 deferred() 替代 column_property()

from sqlalchemy.orm import deferred
class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    firstname: Mapped[str] = mapped_column()
    lastname: Mapped[str] = mapped_column()
    fullname: Mapped[str] = deferred(firstname + " " + lastname)

请参阅

使用 column_property - 在 SQL 表达式作为映射属性 部分中

应用 Imperative 表列的加载、持久化和映射选项 - 在 声明式表配置 部分中

使用undefer()“急切”加载延迟列

使用默认延迟列配置的映射上的列,undefer()选项将导致通常延迟的任何列被解除延迟,即,与映射的所有其他列一起前端加载。例如,我们可以将undefer()应用于前一映射中指定为延迟的Book.summary列:

>>> from sqlalchemy.orm import undefer
>>> book = session.scalar(select(Book).where(Book.id == 2).options(undefer(Book.summary)))
SELECT  book.id,  book.owner_id,  book.title,  book.summary
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

Book.summary列现在已经被急切加载,可以在不发出额外 SQL 的情况下访问:

>>> print(book.summary)
another long summary

按组加载延迟列

通常,当列被映射为mapped_column(deferred=True)时,当在对象上访问延迟属性时,将发出 SQL 仅加载该特定列,而不加载其他列,即使映射还有其他列也被标记为延迟。在常见情况下,延迟属性是一组应该同时加载的属性的一部分时,而不是为每个属性单独发出 SQL,可以使用mapped_column.deferred_group参数,该参数接受一个任意字符串,该字符串将定义一个通用列组以解除延迟:

>>> 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_group="book_attrs"
...     )
...     cover_photo: Mapped[bytes] = mapped_column(
...         LargeBinary, deferred=True, deferred_group="book_attrs"
...     )
...
...     def __repr__(self) -> str:
...         return f"Book(id={self.id!r}, title={self.title!r})"

使用上述映射,访问summarycover_photo将一次性使用一个 SELECT 语句加载两个列:

>>> book = session.scalar(select(Book).where(Book.id == 2))
SELECT  book.id,  book.owner_id,  book.title
FROM  book
WHERE  book.id  =  ?
[...]  (2,)
>>> img_data, summary = book.cover_photo, book.summary
SELECT  book.summary  AS  book_summary,  book.cover_photo  AS  book_cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

使用undefer_group()按组解除延迟

如果延迟列配置为mapped_column.deferred_group,如前一节介绍的,可以通过指定要急切加载的组的字符串名称来指示整个组的加载:

>>> from sqlalchemy.orm import undefer_group
>>> book = session.scalar(
...     select(Book).where(Book.id == 2).options(undefer_group("book_attrs"))
... )
SELECT  book.id,  book.owner_id,  book.title,  book.summary,  book.cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (2,) 

summarycover_photo都可用,无需额外加载:

>>> img_data, summary = book.cover_photo, book.summary

使用通配符解除延迟加载

大多数 ORM 加载器选项都接受通配符表达式,由 "*" 表示,表示该选项应用于所有相关属性。如果映射具有一系列延迟列,则可以通过指定通配符一次性解除所有这些列的延迟,而无需使用组名:

>>> book = session.scalar(select(Book).where(Book.id == 3).options(undefer("*")))
SELECT  book.id,  book.owner_id,  book.title,  book.summary,  book.cover_photo
FROM  book
WHERE  book.id  =  ?
[...]  (3,) 

配置映射器级别的“raiseload”行为

“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'


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

相关文章
|
4月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档(十九)(2)
SqlAlchemy 2.0 中文文档(十九)
36 2
|
4月前
|
SQL 存储 大数据
SqlAlchemy 2.0 中文文档(十八)(1)
SqlAlchemy 2.0 中文文档(十八)
31 1
|
4月前
|
SQL 测试技术 Go
SqlAlchemy 2.0 中文文档(十八)(2)
SqlAlchemy 2.0 中文文档(十八)
23 1
|
4月前
|
SQL 测试技术 Go
SqlAlchemy 2.0 中文文档(十八)(5)
SqlAlchemy 2.0 中文文档(十八)
28 1
|
4月前
|
SQL 测试技术 Go
SqlAlchemy 2.0 中文文档(十八)(3)
SqlAlchemy 2.0 中文文档(十八)
24 1
|
4月前
|
SQL Java Go
SqlAlchemy 2.0 中文文档(十九)(1)
SqlAlchemy 2.0 中文文档(十九)
32 1
|
4月前
|
SQL 测试技术 知识图谱
SqlAlchemy 2.0 中文文档(十五)(3)
SqlAlchemy 2.0 中文文档(十五)
33 1
|
4月前
|
SQL 测试技术 API
SqlAlchemy 2.0 中文文档(二十)(4)
SqlAlchemy 2.0 中文文档(二十)
30 1
|
4月前
|
SQL 存储 测试技术
SqlAlchemy 2.0 中文文档(二十)(3)
SqlAlchemy 2.0 中文文档(二十)
27 1
|
4月前
|
SQL 缓存 API
SqlAlchemy 2.0 中文文档(二十)(5)
SqlAlchemy 2.0 中文文档(二十)
26 1