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

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

身份令牌

深度炼金术

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

“身份令牌”是一个任意值，可以与新加载对象的 identity key  相关联。此元素首先存在以支持执行按行“sharding”的扩展，其中对象可以从特定数据库表的任何数量的副本中加载，尽管它们具有重叠的主键值。  “身份令牌”的主要消费者是 Horizontal Sharding  扩展，它提供了一种在特定数据库表的多个“shards”之间持久化对象的通用框架。

identity_token执行选项可以根据每个查询直接影响此令牌。直接使用它，可以填充一个Session的多个对象实例，这些对象具有相同的主键和来源表，但具有不同的“身份”。

其中一个示例是使用 Schema Names 的翻译功能来填充一个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 对象是独立的。如果我们想要在一个事务中持久化这两个对象，我们需要使用 水平分片 扩展来执行此操作。

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

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

上述逻辑在使用 水平分片 扩展时会自动进行。

从版本 2.0.0rc1 开始新增： - 添加了 identity_token ORM 层执行选项。

另请参阅

水平分片 - 在 ORM 示例 部分。查看脚本 separate_schema_translates.py，演示了使用完整分片 API 的上述用例。

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

select() 构造，以及 insert()、update() 和 delete() 构造（对于后两个 DML 构造，在 SQLAlchemy 1.4.33 中），都支持检查创建这些语句的实体，以及在结果集中返回的列和数据类型的能力。

对于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'>}]

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

>>> 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，它返回有关主要 ORM 实体和数据库表的信息，该信息会受到 DML 构造的影响：

>>> 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.orm.AliasedClass (sqlalchemy.inspection.Inspectable, sqlalchemy.orm.ORMColumnsClauseRole)

class sqlalchemy.orm.util.AliasedInsp

为 AliasedClass 对象提供检查接口。

给定AliasedClass，使用inspect()函数返回AliasedInsp对象：

from sqlalchemy import inspect
from sqlalchemy.orm import aliased
my_alias = aliased(MyMappedClass)
insp = inspect(my_alias)

AliasedInsp的属性包括：

• entity - 表示的AliasedClass。
  
• mapper - 映射底层类的Mapper。
  
• selectable - 最终表示别名Table或Select构造的Alias构造。
  
• name - 别名的名称。当从Query返回结果元组时，也用作属性名称。
  
• with_polymorphic_mappers - 指示选择构造中表示所有这些映射的Mapper对象集合，用于AliasedClass。
  
• polymorphic_on - 用作多态加载的“鉴别器”的备用列或 SQL 表达式。
  

另请参阅

运行时检查 API

类签名

类 sqlalchemy.orm.AliasedInsp (sqlalchemy.orm.ORMEntityColumnsClauseRole, sqlalchemy.orm.ORMFromClauseRole, sqlalchemy.sql.cache_key.HasCacheKey, sqlalchemy.orm.base.InspectionAttr, sqlalchemy.util.langhelpers.MemoizedSlots, sqlalchemy.inspection.Inspectable, typing.Generic)

class sqlalchemy.orm.Bundle

由一个命名空间下的Query返回的 SQL 表达式分组。

Bundle基本上允许通过简单的子类化来嵌套由基于列的Query对象返回的基于元组的结果。它还可以通过简单的子类化来扩展，其中要重写的主要功能是如何返回表达式集，允许进行后处理以及自定义返回类型，而无需涉及 ORM 身份映射的类。

另请参阅

使用 Bundle 对选定的属性进行分组

成员

init(), c, columns, create_row_processor(), is_aliased_class, is_bundle, is_clause_element, is_mapper, label(), single_entity

类签名

class sqlalchemy.orm.Bundle (sqlalchemy.orm.ORMColumnsClauseRole, sqlalchemy.sql.annotation.SupportsCloneAnnotations, sqlalchemy.sql.cache_key.MemoizedHasCacheKey, sqlalchemy.inspection.Inspectable, sqlalchemy.orm.base.InspectionAttr)

method __init__(name: str, *exprs: _ColumnExpressionArgument[Any], **kw: Any)

构建一个新的 Bundle。

例如：

bn = Bundle("mybundle", MyClass.x, MyClass.y)
for row in session.query(bn).filter(
 bn.c.x == 5).filter(bn.c.y == 4):
 print(row.mybundle.x, row.mybundle.y)

参数：

• name – bundle 的名称。
  
• *exprs – 组成 bundle 的列或 SQL 表达式。
  
• single_entity=False – 如果为 True，则此 Bundle 的行可以作为“单个实体”返回，方式与映射实体相同，不在任何封闭元组之外。
  

attribute c: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]

Bundle.columns 的别名。

attribute columns: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]

此 Bundle 所引用的 SQL 表达式的命名空间。

例如：

bn = Bundle("mybundle", MyClass.x, MyClass.y)
q = sess.query(bn).filter(bn.c.x == 5)

还支持 bundle 的嵌套：

b1 = Bundle("b1",
 Bundle('b2', MyClass.a, MyClass.b),
 Bundle('b3', MyClass.x, MyClass.y)
 )
q = sess.query(b1).filter(
 b1.c.b2.c.a == 5).filter(b1.c.b3.c.y == 9)

另请参阅

Bundle.c

method create_row_processor(query: Select[Any], procs: Sequence[Callable[[Row[Any]], Any]], labels: Sequence[str]) → Callable[[Row[Any]], Any]

生成此 Bundle 的“行处理”函数。

可以被子类重写以在获取结果时提供自定义行为。该方法在查询执行时传递了语句对象和一组“行处理器”函数；这些处理器函数在给定结果行时将返回单个属性值，然后可以将其适应为任何类型的返回数据结构。

下面的示例说明了将通常的 Row 返回结构替换为直接的 Python 字典：

from sqlalchemy.orm import Bundle
class DictBundle(Bundle):
 def create_row_processor(self, query, procs, labels):
 'Override create_row_processor to return values as
 dictionaries'
 def proc(row):
 return dict(
 zip(labels, (proc(row) for proc in procs))
 )
 return proc

上述 Bundle 的结果将返回字典值：

bn = DictBundle('mybundle', MyClass.data1, MyClass.data2)
for row in session.execute(select(bn)).where(bn.c.data1 == 'd1'):
 print(row.mybundle['data1'], row.mybundle['data2'])

attribute is_aliased_class = False

如果此对象是 AliasedClass 的实例，则为 True。

attribute is_bundle = True

如果此对象是 Bundle 的实例，则为 True。

attribute is_clause_element = False

如果此对象是 ClauseElement 的实例，则为 True。

attribute is_mapper = False

如果此对象是 Mapper 的实例，则为 True。

method label(name)

提供此 Bundle 的副本并传递一个新标签。

attribute single_entity = False

如果为 True，则单个 Bundle 的查询将作为单个实体返回，而不是作为键元组中的元素。

function sqlalchemy.orm.with_loader_criteria(entity_or_base: _EntityType[Any], where_criteria: _ColumnExpressionArgument[bool] | Callable[[Any], _ColumnExpressionArgument[bool]], loader_only: bool = False, include_aliases: bool = False, propagate_to_loaders: bool = True, track_closure_variables: bool = True) → LoaderCriteriaOption

为特定实体的所有出现添加额外的 WHERE 条件。

在 1.4 版本中新增。

with_loader_criteria()选项旨在向查询中的特定实体添加限制条件，全局地应用于实体在 SELECT 查询中的出现以及任何子查询、连接条件和关系加载中，包括急切加载和延迟加载器，而无需在查询的任何特定部分指定它。渲染逻辑使用与单表继承相同的系统，以确保某个鉴别器应用于表。

例如，使用 2.0 风格的查询，我们可以限制User.addresses集合的加载方式，无论使用何种加载方式：

from sqlalchemy.orm import with_loader_criteria
stmt = select(User).options(
 selectinload(User.addresses),
 with_loader_criteria(Address, Address.email_address != 'foo'))
)

上述对User.addresses的“selectinload”将把给定的过滤条件应用于 WHERE 子句。

另一个例子，过滤将应用于连接的 ON 子句，在这个例子中使用 1.x 风格的查询：

q = session.query(User).outerjoin(User.addresses).options(
 with_loader_criteria(Address, Address.email_address != 'foo'))
)

with_loader_criteria()的主要目的是在SessionEvents.do_orm_execute()事件处理程序中使用它，以确保以某种方式过滤特定实体的所有出现，例如过滤访问控制角色。它还可以用于应用关系加载的条件。在下面的例子中，我们可以对特定Session发出的所有查询应用一定的规则：

session = Session(bind=engine)
@event.listens_for("do_orm_execute", session)
def _add_filtering_criteria(execute_state):
 if (
 execute_state.is_select
 and not execute_state.is_column_load
 and not execute_state.is_relationship_load
 ):
 execute_state.statement = execute_state.statement.options(
 with_loader_criteria(
 SecurityRole,
 lambda cls: cls.role.in_(['some_role']),
 include_aliases=True
 )
 )

在上面的例子中，SessionEvents.do_orm_execute()事件将拦截使用Session发出的所有查询。对于那些是 SELECT 语句且不是属性或关系加载的查询，将为查询添加自定义的with_loader_criteria()选项。with_loader_criteria()选项将用于给定语句，并将自动传播到所有从此查询派生的关系加载。

给定的 criteria 参数是一个接受cls参数的lambda。给定的类将扩展以包括所有映射的子类，本身不必是一个映射的类。

提示

当与with_loader_criteria()选项一起使用时，需要注意with_loader_criteria()仅影响查询中确定渲染的 SQL 的部分，即 WHERE 和 FROM 子句。contains_eager()选项不会影响 SELECT 语句的渲染，除了列子句外的其他部分，因此与with_loader_criteria()选项没有任何交互。然而，“工作”的方式是contains_eager()旨在与已经以某种方式从其他实体进行选择的查询一起使用，而with_loader_criteria()可以应用其额外的条件。

在下面的示例中，假设一个映射关系为A -> A.bs -> B，给定的with_loader_criteria()选项将影响 JOIN 的渲染方式：

stmt = select(A).join(A.bs).options(
 contains_eager(A.bs),
 with_loader_criteria(B, B.flag == 1)
)

在上面的例子中，给定的with_loader_criteria()选项将影响由.join(A.bs)指定的 JOIN 的 ON 子句，因此会按预期应用。contains_eager()选项会导致B的列被添加到列子句中：

SELECT
 b.id, b.a_id, b.data, b.flag,
 a.id AS id_1,
 a.data AS data_1
FROM a JOIN b ON a.id = b.a_id AND b.flag = :flag_1

在上述语句中使用contains_eager()选项对with_loader_criteria()选项的行为没有影响。如果省略contains_eager()选项，则 SQL 将与 FROM 和 WHERE 子句相关，而with_loader_criteria()将继续将其条件添加到 JOIN 的 ON 子句中。添加contains_eager()仅会影响列子句，即会添加对b的额外列，然后 ORM 会使用这些列来生成B实例。

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