SqlAlchemy 2.0 中文文档（二十）（2）

SqlAlchemy 2.0 中文文档（二十）（1）https://developer.aliyun.com/article/1560345

ORM Loader 选项

Loader 选项是对象，当传递给 Select.options() 方法时，影响了 Select 对象或类似的 SQL 结构的列和关系属性的加载。大多数 loader 选项都来自 Load 层次结构。有关使用 loader 选项的完整概述，请参阅下面的链接部分。

另请参阅

• 列加载选项 - 详细介绍了影响如何加载列和 SQL 表达式映射属性的映射和加载选项
  
• 关系加载技术 - 详细介绍了影响如何加载 relationship() 映射属性的关系和加载选项
  

ORM 执行选项

ORM 级别的执行选项是关键字选项，可以通过Session.execute.execution_options参数与语句执行关联，该参数是由Session方法（如Session.execute()和Session.scalars()）接受的字典参数，或者直接通过Executable.execution_options()方法将它们与要调用的语句直接关联，该方法将它们作为任意关键字参数接受。

ORM 级别的选项与Connection.execution_options()中记录的核心级别执行选项不同。需要注意的是，下面讨论的 ORM 选项与核心级别方法Connection.execution_options()或Engine.execution_options() 不兼容；即使Engine或Connection与正在使用的Session相关联，这些选项在此级别将被忽略。

在本节中，将演示Executable.execution_options()方法的样式示例。

刷新现有数据

populate_existing执行选项确保对于加载的所有行，Session中对应的实例将被完全刷新 - 擦除对象中的任何现有数据（包括待定更改）并用从结果加载的数据替换。

示例用法如下：

>>> stmt = select(User).execution_options(populate_existing=True)
>>> result = session.execute(stmt)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
... 

通常，ORM 对象只加载一次，如果它们与后续结果行中的主键匹配，则不会将该行应用于对象。这既是为了保留对象上未决的未刷新更改，也是为了避免刷新已经存在的数据的开销和复杂性。 Session 假定高度隔离的事务的默认工作模型，并且在预期事务内部的数据发生更改的程度上，会使用明确的步骤来处理那些使用情况，而不是正在进行的本地更改。

使用 populate_existing，可以刷新与查询匹配的任何一组对象，并且还可以控制关系加载器选项。例如，刷新一个实例同时也刷新相关的一组对象：

stmt = (
    select(User)
    .where(User.name.in_(names))
    .execution_options(populate_existing=True)
    .options(selectinload(User.addresses))
)
# will refresh all matching User objects as well as the related
# Address objects
users = session.execute(stmt).scalars().all()

populate_existing 的另一个用例是支持各种属性加载功能，这些功能可以根据每个查询的情况改变属性的加载方式。适用于此选项的选项包括：

• with_expression() 选项
  
• PropComparator.and_() 方法可以修改加载器策略加载的内容
  
• contains_eager() 选项
  
• with_loader_criteria() 选项
  
• load_only() 选项以选择要刷新的属性
  

populate_existing 执行选项等效于 1.x 风格 ORM 查询中的 Query.populate_existing() 方法。

另请参阅

我正在使用我的 Session 重新加载数据，但它没有看到我在其他地方提交的更改 - 在常见问题解答中

刷新/过期 - 在 ORM Session 文档中 ### 自动刷新

当传递此选项为 False 时，将导致 Session 不调用“自动刷新”步骤。这相当于使用 Session.no_autoflush 上下文管理器来禁用自动刷新：

>>> stmt = select(User).execution_options(autoflush=False)
>>> session.execute(stmt)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
... 

此选项还可用于启用 ORM 的 Update 和 Delete 查询。

autoflush 执行选项相当于 1.x 风格 ORM 查询中的 Query.autoflush() 方法。

另请参阅

刷新 ### 使用 Yield Per 获取大型结果集

yield_per 执行选项是一个整数值，它将导致 Result 一次仅缓冲有限数量的行和/或 ORM 对象，然后将数据提供给客户端。

通常，ORM 会立即获取所有行，为每个构造 ORM 对象，并将这些对象组装到一个单一的缓冲区中，然后将该缓冲区作为行的来源传递给 Result 对象以返回。此行为的理由是允许正确处理诸如联接急加载、结果唯一化以及依赖于标识映射在每个对象在被提取时保持一致状态的结果处理逻辑等功能的情况。

yield_per 选项的目的是更改此行为，使得 ORM 结果集针对迭代大型结果集（例如 > 10K 行）进行了优化，用户已确定上述模式不适用的情况。当使用 yield_per 时，ORM 将会将 ORM 结果批量成子集合，并在迭代 Result 对象时单独从每个子集合中产生行，这样 Python 解释器就不需要声明非常大的内存区域，这既耗时又导致内存使用过多。该选项既影响数据库游标的使用方式，也影响 ORM 构造要传递给 Result 的行和对象的方式。

提示

由上可见，Result 必须以可迭代的方式消耗，即使用诸如 for row in result 的迭代或使用部分行方法，如 Result.fetchmany() 或 Result.partitions()。调用 Result.all() 将失去使用 yield_per 的目的。

使用yield_per等同于同时利用Connection.execution_options.stream_results执行选项，该选项在支持的情况下选择使用后端的服务器端游标，并且在返回的Result对象上使用Result.yield_per()方法，该方法建立了要获取的行的固定大小以及一次构建多少个 ORM 对象的相应限制。

提示

yield_per现在也作为 Core 执行选项可用，详细描述在使用服务器端游标（即流式结果）。本节详细介绍了将yield_per作为 ORM Session的执行选项的用法。该选项在两种情境下尽可能相似地行为。

在与 ORM 一起使用时，yield_per必须通过给定语句上的Executable.execution_options()方法或通过将其传递给Session.execute()或其他类似Session方法的Session.execute.execution_options参数来建立。获取 ORM 对象的典型用法如下所示：

>>> stmt = select(User).execution_options(yield_per=10)
>>> for user_obj in session.scalars(stmt):
...     print(user_obj)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
...
>>> # ... rows continue ...

上述代码等同于下面的示例，该示例使用Connection.execution_options.stream_results和Connection.execution_options.max_row_buffer Core 级别的执行选项，与Result的Result.yield_per()方法一起使用：

# equivalent code
>>> stmt = select(User).execution_options(stream_results=True, max_row_buffer=10)
>>> for user_obj in session.scalars(stmt).yield_per(10):
...     print(user_obj)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
...
>>> # ... rows continue ...

yield_per通常与Result.partitions()方法结合使用，该方法将迭代分组分区中的行。每个分区的大小默认为传递给yield_per的整数值，如下例所示：

>>> stmt = select(User).execution_options(yield_per=10)
>>> for partition in session.scalars(stmt).partitions():
...     for user_obj in partition:
...         print(user_obj)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
...
>>> # ... rows continue ...

当使用集合时，yield_per执行选项与“子查询”急加载或“连接”急加载不兼容。对于“select in”急加载，只要数据库驱动程序支持多个独立游标，它就可能与之兼容。

另外，yield_per执行选项与Result.unique()方法不兼容；由于此方法依赖于存储所有行的完整标识集，它必然会破坏使用yield_per的目的，即处理任意数量的行。

自 1.4.6 版本更改：当从使用Result对象获取 ORM 行时，如果同时使用了Result.unique()过滤器和yield_per执行选项，则会引发异常。

在使用旧版 Query 对象和 1.x 样式 ORM 时，Query.yield_per() 方法的结果与yield_per执行选项的结果相同。

另请参阅

使用服务器端游标（又名流式结果） ### 标识令牌

深度炼金术

此选项是一个高级特性，主要用于与水平分片扩展一起使用。对于从不同“分片”或分区加载具有相同主键的对象的典型情况，请首先考虑每个分片使用单独的Session对象。

“标识令牌”是一个任意值，可以与新加载对象的标识键相关联。这个元素首先存在于支持每行“分片”的扩展中，其中对象可以从特定数据库表的任意数量的副本中加载，尽管这些副本具有重叠的主键值。  “标识令牌”的主要消费者是 Horizontal Sharding  扩展，它提供了一个��用框架，用于在特定数据库表的多个“分片”之间持久化对象。

identity_token执行选项可以在每个查询基础上直接影响此令牌的使用。直接使用它，可以将一个对象的多个实例填充到Session中，这些实例具有相同的主键和源表，但具有不同的“标识”。

一个这样的例子是使用 Schema 名称翻译功能，该功能可以影响查询范围内的模式选择，从而将来自不同模式的同名表中的对象填充到Session中。给定一个映射如下：

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(DeclarativeBase):
    pass
class MyTable(Base):
    __tablename__ = "my_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

上述类的默认“模式”名称为None，意味着不会将模式限定写入 SQL 语句中。然而，如果我们利用Connection.execution_options.schema_translate_map，将None映射到替代模式，我们可以将MyTable的实例放入两个不同的模式中：

engine = create_engine(
    "postgresql+psycopg://scott:tiger@localhost/test",
)
with Session(
    engine.execution_options(schema_translate_map={None: "test_schema"})
) as sess:
    sess.add(MyTable(name="this is schema one"))
    sess.commit()
with Session(
    engine.execution_options(schema_translate_map={None: "test_schema_2"})
) as sess:
    sess.add(MyTable(name="this is schema two"))
    sess.commit()

上述两个块每次创建一个与不同模式转换映射相关联的Session对象，并将MyTable的实例持久化到test_schema.my_table和test_schema_2.my_table中。

上述Session对象是独立的。如果我们想要在一个事务中持久化这两个对象，我们需要使用 Horizontal Sharding 扩展来实现。

然而，我们可以在一个会话中演示查询这些对象的方法如下：

with Session(engine) as sess:
    obj1 = sess.scalar(
        select(MyTable)
        .where(MyTable.id == 1)
        .execution_options(
            schema_translate_map={None: "test_schema"},
            identity_token="test_schema",
        )
    )
    obj2 = sess.scalar(
        select(MyTable)
        .where(MyTable.id == 1)
        .execution_options(
            schema_translate_map={None: "test_schema_2"},
            identity_token="test_schema_2",
        )
    )

obj1和obj2彼此不同。然而，它们都指向MyTable类的主键 id 1，但是它们是不同的。这就是identity_token发挥作用的地方，我们可以在每个对象的检查中看到它，在那里我们查看InstanceState.key以查看这两个不同的标识令牌：

>>> from sqlalchemy import inspect
>>> inspect(obj1).key
(<class '__main__.MyTable'>, (1,), 'test_schema')
>>> inspect(obj2).key
(<class '__main__.MyTable'>, (1,), 'test_schema_2')

当使用 Horizontal Sharding 扩展时，上述逻辑会自动发生。

2.0.0rc1 版本中的新功能：- 添加了identity_tokenORM 级别执行选项。

另请参阅

水平分片 - 在 ORM 示例 部分。请参阅脚本 separate_schema_translates.py，了解使用完整分片 API 进行上述用例演示。

检查启用 ORM 的 SELECT 和 DML 语句中的实体和列

select() 结构以及 insert()、update() 和 delete() 结构（自 SQLAlchemy 1.4.33 起，对于后三个 DML 结构）都支持检查这些语句所针对的实体，以及结果集中将返回的列和数据类型。

对于 Select 对象，此信息可以从 Select.column_descriptions 属性获取。该属性的操作方式与传统的 Query.column_descriptions 属性相同。返回的格式是一个字典列表：

>>> from pprint import pprint
>>> user_alias = aliased(User, name="user2")
>>> stmt = select(User, User.id, user_alias)
>>> pprint(stmt.column_descriptions)
[{'aliased': False,
 'entity': <class 'User'>,
 'expr': <class 'User'>,
 'name': 'User',
 'type': <class 'User'>},
 {'aliased': False,
 'entity': <class 'User'>,
 'expr': <....InstrumentedAttribute object at ...>,
 'name': 'id',
 'type': Integer()},
 {'aliased': True,
 'entity': <AliasedClass ...; User>,
 'expr': <AliasedClass ...; User>,
 'name': 'user2',
 'type': <class 'User'>}]

当与非 ORM 对象（如普通的 Table 或 Column 对象）一起使用 Select.column_descriptions 时，所有情况下都会包含有关返回的各个列的基本信息：

>>> stmt = select(user_table, address_table.c.id)
>>> pprint(stmt.column_descriptions)
[{'expr': Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False),
 'name': 'id',
 'type': Integer()},
 {'expr': Column('name', String(), table=<user_account>, nullable=False),
 'name': 'name',
 'type': String()},
 {'expr': Column('fullname', String(), table=<user_account>),
 'name': 'fullname',
 'type': String()},
 {'expr': Column('id', Integer(), table=<address>, primary_key=True, nullable=False),
 'name': 'id_1',
 'type': Integer()}]

自 1.4.33 版更改：当对未启用 ORM 的 Select 使用 Select.column_descriptions 属性时，现在会返回一个值。之前会引发 NotImplementedError。

对于insert()、update()和delete()构造，存在两个单独的属性。一个是UpdateBase.entity_description，它返回关于 DML 构造将影响的主 ORM 实体和数据库表的信息：

>>> from sqlalchemy import update
>>> stmt = update(User).values(name="somename").returning(User.id)
>>> pprint(stmt.entity_description)
{'entity': <class 'User'>,
 'expr': <class 'User'>,
 'name': 'User',
 'table': Table('user_account', ...),
 'type': <class 'User'>}

提示

UpdateBase.entity_description包括一个条目"table"，实际上是由语句插入、更新或删除的表，这与类可能映射到的 SQL“selectable”不一定相同。例如，在连接表继承场景中，"table"将引用给定实体的本地表。

另一个是UpdateBase.returning_column_descriptions，它以与Select.column_descriptions大致相似的方式提供了与 RETURNING 集合中存在的列有关的信息：

>>> pprint(stmt.returning_column_descriptions)
[{'aliased': False,
 'entity': <class 'User'>,
 'expr': <sqlalchemy.orm.attributes.InstrumentedAttribute ...>,
 'name': 'id',
 'type': Integer()}]

新版本 1.4.33 中增加了UpdateBase.entity_description和UpdateBase.returning_column_descriptions属性。 #### 额外的 ORM API 构造

function sqlalchemy.orm.aliased(element: _EntityType[_O] | FromClause, alias: FromClause | None = None, name: str | None = None, flat: bool = False, adapt_on_names: bool = False) → AliasedClass[_O] | FromClause | AliasedType[_O]

创建给定元素的别名，通常是AliasedClass实例。

例如：

my_alias = aliased(MyClass)
stmt = select(MyClass, my_alias).filter(MyClass.id > my_alias.id)
result = session.execute(stmt)

aliased()函数用于创建一个映射类到新的可选项的临时映射。默认情况下，通过通常的映射可选项（通常是Table）使用FromClause.alias()方法生成可选项。但是，aliased()也可以用于将类链接到新的select()语句。此外，with_polymorphic()函数是aliased()的变体，旨在指定所谓的“多态可选项”，该可选项对应于一次性联接继承子类的联合。

为了方便起见，aliased()函数也接受普通的FromClause构造，比如Table或select()构造。在这些情况下，对象会调用FromClause.alias()方法，并返回新的Alias对象。在这种情况下，返回的Alias不会被 ORM 映射。

另请参阅

ORM 实体别名 - 在 SQLAlchemy 统一教程 中

选择 ORM 别名 - 在 ORM 查询指南 中

参数：

• element – 要别名的元素。通常是一个映射类，但为了方便，也可以是一个FromClause元素。
  
• alias – 可选的可选择单元，用于将元素映射到。通常用于将对象链接到子查询，并且应该是一个别名选择结构，就像从Query.subquery()方法或Select.subquery()或Select.alias()方法生成的那样select()结构。
  
• name – 如果未由alias参数指定，则使用的可选字符串名称。名称，除其他外，形成了通过Query对象返回的元组访问的属性名称。在创建Join对象的别名时不支持。
  
• flat – 布尔值，将传递到FromClause.alias()调用，以便Join对象的别名将别名加入到连接内的单个表，而不是创建子查询。这通常由所有现代数据库支持，关于右嵌套连接，通常生成更有效的查询。
  
• adapt_on_names –
  如果为 True，则在将 ORM 实体的映射列映射到给定可选择的列时将使用更自由的“匹配” - 如果给定的可选择否则没有与实体上的列对应的列，则将执行基于名称的匹配。这种用例是当将实体与某个派生的可选择相关联时，例如使用聚合函数的可选择：

class UnitPrice(Base):
 __tablename__ = 'unit_price'
 ...
 unit_id = Column(Integer)
 price = Column(Numeric)
aggregated_unit_price = Session.query(
 func.sum(UnitPrice.price).label('price')
 ).group_by(UnitPrice.unit_id).subquery()
aggregated_unit_price = aliased(UnitPrice,
 alias=aggregated_unit_price, adapt_on_names=True)

• 在上面，对aggregated_unit_price上的函数引用.price将返回func.sum(UnitPrice.price).label('price')列，因为它根据名称“price”进行匹配。通常，“price”函数不会与实际的UnitPrice.price列有任何“列对应”，因为它不是原始列的代理。

class sqlalchemy.orm.util.AliasedClass

表示与查询一起使用的映射类的“别名”形式。

ORM 中 alias() 构造的等效对象，该对象使用 __getattr__ 方案模仿映射类，并维护对真实 Alias 对象的引用。

AliasedClass 的一个主要目的是在 ORM 生成的 SQL 语句中作为一个替代，以便一个现有的映射实体可以在多个上下文中使用。一个简单的例子：

# find all pairs of users with the same name
user_alias = aliased(User)
session.query(User, user_alias).\
 join((user_alias, User.id > user_alias.id)).\
 filter(User.name == user_alias.name)

AliasedClass 还能够将现有的映射类映射到一个全新的可选择项，前提是该可选择项与现有的映射可选择项兼容，并且还可以在映射中配置为 relationship() 的目标。请参见下面的链接以获取示例。

AliasedClass 对象通常使用 aliased() 函数构造。在使用 with_polymorphic() 函数时，还会进行额外配置。

结果对象是 AliasedClass 的一个实例。该对象实现了一个属性方案，产生与原始映射类相同的属性和方法接口，允许 AliasedClass 与在原始类上工作的任何属性技术兼容，包括混合属性（参见 混合属性）。

AliasedClass 可以通过 inspect() 进行检查，以获取其底层的 Mapper、别名可选择项和其他信息：

from sqlalchemy import inspect
my_alias = aliased(MyClass)
insp = inspect(my_alias)

结果检查对象是 AliasedInsp 的一个实例。

另请参阅

aliased()

with_polymorphic()

与别名类的关系

使用窗口函数限制行关系

SqlAlchemy 2.0 中文文档（二十）（3）https://developer.aliyun.com/article/1560350