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
的查询将自动不包括 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,)
与所有延迟加载属性一样,当首次访问加载的对象上的延迟加载属性时,默认行为是它们将 延迟加载 它们的值:
>>> 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})"
使用上述映射,访问summary
或cover_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,)
summary
和cover_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})"
使用上述映射,访问summary
或cover_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,)
summary
和cover_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