标签：__ 十七 sqlalchemy self py hybrid SqlAlchemy 文档 property

SqlAlchemy 2.0 中文文档（十七）

原文：docs.sqlalchemy.org/en/20/contents.html

排序列表

原文：docs.sqlalchemy.org/en/20/orm/extensions/orderinglist.html

一个管理包含元素的索引/位置信息的自定义列表。

作者：

Jason Kirtland

orderinglist是一个用于可变有序关系的辅助程序。它将拦截对由relationship()管理的集合执行的列表操作，并自动将列表位置的更改同步到目标标量属性。

示例：一个slide表，其中每行引用相关bullet表中的零个或多个条目。幻灯片中的子弹根据bullet表中position列的值按顺序显示。当内存中重新排序条目时，position属性的值应更新以反映新的排序顺序：

Base = declarative_base()

class Slide(Base):
    __tablename__ = 'slide'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    bullets = relationship("Bullet", order_by="Bullet.position")

class Bullet(Base):
    __tablename__ = 'bullet'
    id = Column(Integer, primary_key=True)
    slide_id = Column(Integer, ForeignKey('slide.id'))
    position = Column(Integer)
    text = Column(String)

标准关系映射将在每个Slide上产生一个类似列表的属性，其中包含所有相关的Bullet对象，但无法自动处理顺序变化。将Bullet附加到Slide.bullets时，Bullet.position属性将保持未设置状态，直到手动分配。当Bullet插入列表中间时，后续的Bullet对象也需要重新编号。

OrderingList对象自动化此任务，管理集合中所有Bullet对象的position属性。它是使用ordering_list()工厂构建的：

from sqlalchemy.ext.orderinglist import ordering_list

Base = declarative_base()

class Slide(Base):
    __tablename__ = 'slide'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    bullets = relationship("Bullet", order_by="Bullet.position",
                            collection_class=ordering_list('position'))

class Bullet(Base):
    __tablename__ = 'bullet'
    id = Column(Integer, primary_key=True)
    slide_id = Column(Integer, ForeignKey('slide.id'))
    position = Column(Integer)
    text = Column(String)

使用上述映射，Bullet.position属性被管理：

s = Slide()
s.bullets.append(Bullet())
s.bullets.append(Bullet())
s.bullets[1].position
>>> 1
s.bullets.insert(1, Bullet())
s.bullets[2].position
>>> 2

OrderingList构造仅适用于对集合的更改，而不是从数据库的初始加载，并要求在加载时对列表进行排序。因此，请确保在针对目标排序属性的relationship()上指定order_by，以便在首次加载时排序正确。

警告

当主键列或唯一列是排序的目标时，OrderingList在功能上提供的功能有限。不支持或存在问题的操作包括：

两个条目必须交换值。在主键或唯一约束的情况下，这不受直接支持，因为这意味着至少需要先暂时删除一行，或者在交换发生时将其更改为第三个中性值。

必须删除一个条目以为新条目腾出位置。SQLAlchemy 的工作单元在单次刷新中执行所有 INSERT 操作，然后再执行 DELETE 操作。在主键的情况下，它将交换相同主键的 INSERT/DELETE 以减轻此限制的影响，但对于唯一列不会发生这种情况。未来的功能将允许“DELETE before INSERT”行为成为可能，从而减轻此限制，但此功能将需要在映射器级别对要以这种方式处理的列集进行显式配置。

ordering_list()接受相关对象的排序属性名称作为参数。默认情况下，对象在ordering_list()中的位置与排序属性同步：索引 0 将获得位置 0，索引 1 位置 1，依此类推。要从 1 或其他整数开始编号，请提供count_from=1。

API 参考

对象名称	描述
count_from_0(index, collection)	编号函数：从 0 开始的连续整数。
count_from_1(index, collection)	编号函数：从 1 开始的连续整数。
count_from_n_factory(start)	编号函数：从任意起始位置开始的连续整数。
ordering_list(attr[, count_from, ordering_func, reorder_on_append])	为映射器定义准备一个`OrderingList`工厂。
OrderingList	一个自定义列表，用于管理其子项的位置信息。

function sqlalchemy.ext.orderinglist.ordering_list(attr: str, count_from: int | None = None, ordering_func: Callable[[int, Sequence[_T]], int] | None = None, reorder_on_append: bool = False) → Callable[[], OrderingList]

准备一个OrderingList工厂，用于在映射器定义中使用。

返回一个适用于作为 Mapper 关系的collection_class选项参数的对象。例如：

from sqlalchemy.ext.orderinglist import ordering_list

class Slide(Base):
    __tablename__ = 'slide'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    bullets = relationship("Bullet", order_by="Bullet.position",
                            collection_class=ordering_list('position'))

参数：

attr – 用于存储和检索排序信息的映射属性的名称
count_from – 设置从count_from开始的基于整数的排序。例如，ordering_list('pos', count_from=1)将在 SQL 中创建一个基于 1 的列表，将值存储在‘pos’列中。如果提供了ordering_func，则会被忽略。

额外的参数传递给OrderingList构造函数。

function sqlalchemy.ext.orderinglist.count_from_0(index, collection)

编号函数：从 0 开始的连续整数。

function sqlalchemy.ext.orderinglist.count_from_1(index, collection)

编号函数：从 1 开始的连续整数。

function sqlalchemy.ext.orderinglist.count_from_n_factory(start)

编号函数：从任意起始位置开始的连续整数。

class sqlalchemy.ext.orderinglist.OrderingList

一个自定义列表，用于管理其子项的位置信息。

OrderingList 对象通常使用 ordering_list() 工厂函数设置，与 relationship() 函数结合使用。

成员

init(), append(), insert(), pop(), remove(), reorder()

类签名

类 sqlalchemy.ext.orderinglist.OrderingList (builtins.list, typing.Generic)

method __init__(ordering_attr: str | None = None, ordering_func: Callable[[int, Sequence[_T]], int] | None = None, reorder_on_append: bool = False)

一个自定义列表，用于管理其子项的位置信息。

OrderingList 是一个 collection_class 列表实现，它将 Python 列表中的位置与映射对象上的位置属性同步。

此实现依赖于列表以正确的顺序开始，因此一定要确保在关系上放置一个 order_by。

参数：

ordering_attr – 存储对象在关系中顺序的属性名称。
ordering_func –

可选。将 Python 列表中的位置映射到存储在 ordering_attr 中的值的函数。返回的值通常（但不必！）是整数。

ordering_func 被调用时具有两个位置参数：列表中元素的索引和列表本身。

如果省略，将使用 Python 列表索引作为属性值。此模块提供了两个基本的预定义编号函数：count_from_0 和 count_from_1。有关诸如步进编号、字母编号和斐波那契编号等更奇特的示例，请参阅单元测试。
reorder_on_append –

默认为 False。当追加一个具有现有（非 None）排序值的对象时，该值将保持不变，除非 reorder_on_append 为真。这是一种优化，可以避免各种危险的意外数据库写入。

SQLAlchemy 将在对象加载时通过 append() 将实例添加到列表中。如果由于某种原因数据库的结果集跳过了排序步骤（例如，行 '1' 缺失，但你得到了 '2'、'3' 和 '4'），那么 reorder_on_append=True 将立即重新编号为 '1'、'2'、'3'。如果有多个会话进行更改，其中任何一个会话恰巧加载了这个集合，即使是临时加载，所有会话都会尝试在它们的提交中“清理”编号，可能会导致除一个之外的所有会话都以并发修改错误失败。

建议保留默认值为 False，如果你在之前已经排序过的实例上进行append()操作，或者在手动执行 SQL 操作后进行一些清理工作时，只需调用reorder()即可。

method append(entity)

将对象追加到列表的末尾。

method insert(index, entity)

在索引之前插入对象。

method pop(index=-1)

移除并返回索引处的项目（默认为最后一个）。

如果列表为空或索引超出范围，则引发 IndexError。

method remove(entity)

移除第一次出现的值。

如果值不存在则引发 ValueError。

method reorder() → None

同步整个集合的排序。

扫描列表并确保每个对象设置了准确的排序信息。

API 参考

对象名称	描述
count_from_0(index, collection)	编号函数：从 0 开始的连续整数。
count_from_1(index, collection)	编号函数：从 1 开始的连续整数。
count_from_n_factory(start)	编号函数：从任意起始值开始的连续整数。
ordering_list(attr[, count_from, ordering_func, reorder_on_append])	为映射器定义准备一个`OrderingList`工厂。
OrderingList	一个自定义列表，管理其子项的位置信息。

function sqlalchemy.ext.orderinglist.ordering_list(attr: str, count_from: int | None = None, ordering_func: Callable[[int, Sequence[_T]], int] | None = None, reorder_on_append: bool = False) → Callable[[], OrderingList]

为映射器定义准备一个OrderingList工厂。

返回一个适合用作 Mapper 关系的collection_class选项参数的对象。例如：

from sqlalchemy.ext.orderinglist import ordering_list

class Slide(Base):
    __tablename__ = 'slide'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    bullets = relationship("Bullet", order_by="Bullet.position",
                            collection_class=ordering_list('position'))

参数：

attr – 用于存储和检索排序信息的映射属性的名称
count_from – 设置基于整数的排序，从count_from开始。例如，ordering_list('pos', count_from=1)将在 SQL 中创建一个以 1 为基础的列表，在‘pos’列中存储值。如果提供了ordering_func，则忽略。

额外的参数将传递给OrderingList构造函数。

function sqlalchemy.ext.orderinglist.count_from_0(index, collection)

编号函数：从 0 开始的连续整数。

function sqlalchemy.ext.orderinglist.count_from_1(index, collection)

编号函数：从 1 开始的连续整数。

function sqlalchemy.ext.orderinglist.count_from_n_factory(start)

编号函数：从任意起始值开始的连续整数。

class sqlalchemy.ext.orderinglist.OrderingList

一个自定义列表，管理其子项的位置信息。

OrderingList对象通常使用与relationship()函数配合使用的ordering_list()工厂函数设置。

成员

init(), append(), insert(), pop(), remove(), reorder()

类签名

类 sqlalchemy.ext.orderinglist.OrderingList (builtins.list, typing.Generic)

method __init__(ordering_attr: str | None = None, ordering_func: Callable[[int, Sequence[_T]], int] | None = None, reorder_on_append: bool = False)

一个自定义列表，用于管理其子项的位置信息。

OrderingList 是一个 collection_class 列表实现，将 Python 列表中的位置与映射对象上的位置属性同步。

此实现依赖于列表以正确顺序开始，因此请务必在关系上放置 order_by。

参数:

ordering_attr – 存储对象在关系中顺序的属性名称。
ordering_func –

可选。将 Python 列表��的位置映射到存储在 ordering_attr 中的值的函数。通常返回的值是整数（但不一定是！）。

ordering_func 被调用时带有两个位置参数：列表中元素的索引和列表本身。

如果省略，则使用 Python 列表索引作为属性值。本模块提供了两个基本的预构建编号函数：count_from_0 和 count_from_1。有关更奇特的示例，如步进编号、字母和斐波那契编号，请参见单元测试。
reorder_on_append –

默认为 False。在附加具有现有（非 None）排序值的对象时，该值将保持不变，除非 reorder_on_append 为 true。这是一种优化，可避免各种危险的意外数据库写入。

当您的对象加载时，SQLAlchemy 将通过 append() 将实例添加到列表中。如果由于某种原因数据库的结果集跳过了排序步骤（例如，行‘1’丢失，但您得到‘2’、‘3’和‘4’），reorder_on_append=True 将立即重新编号项目为‘1’、‘2’、‘3’。如果有多个会话进行更改，其中任何一个碰巧加载此集合，即使是临时加载，所有会话都会尝试在其提交中“清理”编号，可能导致除一个之外的所有会话都因并发修改错误而失败。

建议保持默认值为 False，并在对先前有序实例进行 append() 操作或在手动 sql 操作后进行一些清理时，只调用 reorder()。

method append(entity)

将对象追加到列表末尾。

method insert(index, entity)

在索引之前插入对象。

method pop(index=-1)

移除并返回索引处的项目（默认为最后一个）。

如果列表为空或索引超出范围，则引发 IndexError。

method remove(entity)

移除第一次出现的值。

如果值不存在，则引发 ValueError。

method reorder() → None

同步整个集合的排序。

扫描列表并确保每个对象具有准确的排序信息设置。

水平分片

原文：docs.sqlalchemy.org/en/20/orm/extensions/horizontal_shard.html

水平分片支持。

定义了一个基本的“水平分片”系统，允许会话在多个数据库之间分发查询和持久化操作。

有关用法示例，请参见源分发中包含的水平分片示例。

深度炼金术

水平分片扩展是一个高级功能，涉及复杂的语句 -> 数据库交互以及对非平凡情况使用半公共 API。在使用这种更复杂且 less-production-tested 系统之前，应始终首先考虑更简单的引用多个数据库“分片”的方法，最常见的是每个“分片”使用一个独立的会话。

API 文档

对象名称	描述
set_shard_id	一个加载器选项，用于为语句应用特定的分片 ID 到主查询，以及为其他关系和列加载器。
分片查询	与`分片会话`一起使用的查询类。
分片会话

class sqlalchemy.ext.horizontal_shard.ShardedSession

成员

init(), connection_callable(), get_bind()

类签名

类sqlalchemy.ext.horizontal_shard.ShardedSession (sqlalchemy.orm.session.Session)

method __init__(shard_chooser: ShardChooser, identity_chooser: Optional[IdentityChooser] = None, execute_chooser: Optional[Callable[[ORMExecuteState], Iterable[Any]]] = None, shards: Optional[Dict[str, Any]] = None, query_cls: Type[Query[_T]] = <class 'sqlalchemy.ext.horizontal_shard.ShardedQuery'>, *, id_chooser: Optional[Callable[[Query[_T], Iterable[_T]], Iterable[Any]]] = None, query_chooser: Optional[Callable[[Executable], Iterable[Any]]] = None, **kwargs: Any) → None

构造一个 ShardedSession。

参数：

shard_chooser – 一个可调用对象，传入一个 Mapper、一个映射实例，可能还有一个 SQL 子句，返回一个分片 ID。该 ID 可能基于对象中存在的属性，或者基于某种轮询方案。如果方案基于选择，则应在实例上设置任何状态，以标记它在未来参与该分片。
identity_chooser –

一个可调用对象，传入一个 Mapper 和主键参数，应返回一个主键可能存在的分片 ID 列表。

在 2.0 版本中更改：identity_chooser 参数取代了 id_chooser 参数。
execute_chooser –

对于给定的ORMExecuteState，返回应发出查询的 shard_ids 列表。从所有返回的 shards 中返回的结果将合并到一个列表中。

在 1.4 版本中更改：execute_chooser参数取代了query_chooser参数。
shards – 一个字符串分片名称到Engine对象的字典。

method connection_callable(mapper: Mapper[_T] | None = None, instance: Any | None = None, shard_id: ShardIdentifier | None = None, **kw: Any) → Connection

提供一个用于工作单元刷新过程中使用的Connection。

method get_bind(mapper: _EntityBindKey[_O] | None = None, *, shard_id: ShardIdentifier | None = None, instance: Any | None = None, clause: ClauseElement | None = None, **kw: Any) → _SessionBind

返回此Session绑定的“bind”。

“bind”通常是Engine的实例，除非Session已经明确地直接绑定到Connection的情况。

对于多重绑定或未绑定的Session，使用mapper或clause参数确定要返回的适当绑定。

请注意，当通过 ORM 操作调用Session.get_bind()时，通常会出现“映射器”参数，例如Session.query()中的每个单独的 INSERT/UPDATE/DELETE 操作，在Session.flush()调用中等。

解析顺序为：

如果提供了映射器并且Session.binds存在，则首先基于正在使用的映射器，然后基于正在使用的映射类，然后基于映射类的__mro__中存在的任何基类，从更具体的超类到更一般的超类进行绑定定位。
如果提供了子句并且Session.binds存在，则基于在Session.binds中找到的给定子句中存在的Table对象进行绑定定位。
如果Session.binds存在，则返回该绑定。
如果提供了子句，则尝试返回一个与最终与该子句关联的MetaData绑定。
如果提供了映射器，尝试返回一个与最终与该映射器映射的Table或其他可选择对象关联的MetaData绑定。
找不到绑定时，会引发UnboundExecutionError。

请注意，Session.get_bind() 方法可以在用户定义的 Session 子类上被重写，以提供任何类型的绑定解析方案。请参阅自定义垂直分区中的示例。

参数：

mapper – 可选的映射类或相应的 Mapper 实例。绑定可以首先从与此 Session 关联的“binds”映射中查询 Mapper，其次从 Mapper 映射到的 Table 的 MetaData 中查询绑定。
clause – 一个 ClauseElement（即 select()、text() 等）。如果 mapper 参数不存在或无法生成绑定，则将搜索给定的表达式构造，通常是与绑定的 MetaData 关联的 Table。

另请参阅

分区策略（例如每个会话多个数据库后端）

Session.binds

Session.bind_mapper()

Session.bind_table()

class sqlalchemy.ext.horizontal_shard.set_shard_id

语句的加载器选项，用于将特定的分片 id 应用于主查询，以及额外的关系和列加载器。

set_shard_id 选项可以使用任何可执行语句的 Executable.options() 方法应用：

stmt = (
    select(MyObject).
    where(MyObject.name == 'some name').
    options(set_shard_id("shard1"))
)

上面的语句在调用时将限制为主查询的“shard1”分片标识符，以及所有关系和列加载策略，包括像selectinload()这样的急切加载器，像defer()这样的延迟列加载器，以及惰性关系加载器lazyload()。

这样，set_shard_id选项的范围比在Session.execute.bind_arguments字典中使用“shard_id”参数要广泛得多。

2.0.0 版本中的新功能。

成员

init(), propagate_to_loaders

类签名

类sqlalchemy.ext.horizontal_shard.set_shard_id (sqlalchemy.orm.ORMOption)

method __init__(shard_id: str, propagate_to_loaders: bool = True)

构造一个set_shard_id选项。

参数：

shard_id – 分片标识符
propagate_to_loaders – 如果保持默认值True，则分片选项将适用于诸如lazyload()和defer()之类的惰性加载器；如果为 False，则该选项不会传播到加载的对象。请注意，defer()在任何情况下始终限制为父行的 shard_id，因此该参数仅对lazyload()策略的行为产生净效果。

attribute propagate_to_loaders

如果为 True，则表示此选项应该在“次要”SELECT 语句中传递，这些语句发生在关系惰性加载器以及属性加载/刷新操作中。

class sqlalchemy.ext.horizontal_shard.ShardedQuery

与ShardedSession一起使用的查询类。

遗留特性

ShardedQuery是遗留Query类的子类。 ShardedSession现在通过ShardedSession.execute()方法支持 2.0 风格的执行。

成员

set_shard()

类签名

类sqlalchemy.ext.horizontal_shard.ShardedQuery (sqlalchemy.orm.Query)的构造函数

method set_shard(shard_id: str) → Self

返回一个新的查询，限制为单个分片 ID。

返回的查询的所有后续操作将针对单个分片执行，而不考虑其他状态。

分片 ID 可以传递给 Session.execute() 的 bind_arguments 字典，以进行 2.0 样式的执行：

results = session.execute(
    stmt,
    bind_arguments={"shard_id": "my_shard"}
)

API 文档

对象名称	描述
set_shard_id	用于语句的加载器选项，以将特定的分片 ID 应用于主查询，以及额外的关系和列加载器。
ShardedQuery	与 `ShardedSession` 一起使用的查询类。
ShardedSession

class sqlalchemy.ext.horizontal_shard.ShardedSession

成员

init(), connection_callable(), get_bind()

类签名

类sqlalchemy.ext.horizontal_shard.ShardedSession (sqlalchemy.orm.session.Session)的构造函数

method __init__(shard_chooser: ShardChooser, identity_chooser: Optional[IdentityChooser] = None, execute_chooser: Optional[Callable[[ORMExecuteState], Iterable[Any]]] = None, shards: Optional[Dict[str, Any]] = None, query_cls: Type[Query[_T]] = <class 'sqlalchemy.ext.horizontal_shard.ShardedQuery'>, *, id_chooser: Optional[Callable[[Query[_T], Iterable[_T]], Iterable[Any]]] = None, query_chooser: Optional[Callable[[Executable], Iterable[Any]]] = None, **kwargs: Any) → None

构造一个 ShardedSession。

参数:

shard_chooser – 一个可调用对象，传入 Mapper、映射实例和可能的 SQL 子句，返回一个分片 ID。此 ID 可能基于对象中存在的属性，或者基于某种循环选择方案。如果方案基于选择，则应在实例上设置将来标记其参与该分片的任何状态。
identity_chooser –

一个可调用对象，传入 Mapper 和主键参数，应返回此主键可能存在的分片 ID 列表。

在 2.0 版本中更改：identity_chooser 参数取代了 id_chooser 参数。
execute_chooser –

对于给定的ORMExecuteState，返回应发出查询的分片 ID 列表。返回的所有分片的结果将合并为单个列表。

在 1.4 版本中更改：execute_chooser 参数取代了 query_chooser 参数。
shards – 一个字符串分片名称到Engine 对象的字典。

method connection_callable(mapper: Mapper[_T] | None = None, instance: Any | None = None, shard_id: ShardIdentifier | None = None, **kw: Any) → Connection

提供一个用于工作单元刷新过程中使用的Connection。

method get_bind(mapper: _EntityBindKey[_O] | None = None, *, shard_id: ShardIdentifier | None = None, instance: Any | None = None, clause: ClauseElement | None = None, **kw: Any) → _SessionBind

返回此Session绑定的“绑定”。

“绑定”通常是Engine的一个实例，除非Session已经被显式地直接绑定到Connection。

对于多重绑定或未绑定的Session，将使用mapper或clause参数来确定要返回的适当绑定。

请注意，“mapper”参数通常在通过 ORM 操作调用Session.get_bind()时存在，例如Session.query()中的每个单独的 INSERT/UPDATE/DELETE 操作，在Session.flush()调用等。

解析的顺序如下：

如果给定了映射器并且存在Session.binds，则首先基于正在使用的映射器，然后基于正在使用的映射类，然后基于映射类的__mro__中存在的任何基类，从更具体的超类到更一般的超类。
如果给定了子句并且存在Session.binds，则基于在Session.binds中找到的给定子句中的Table对象定位绑定。
如果存在Session.binds，则返回该绑定。
如果给定了子句，则尝试返回与该子句最终相关联的MetaData相关的绑定。
如果给定了映射器，则尝试返回与最终与该映射器映射的Table或其他可选择对象相关联的MetaData相关的绑定。
找不到绑定时，会引发UnboundExecutionError。

注意，Session.get_bind()方法可以在Session的用户定义子类上被覆盖，以提供任何类型的绑定解析方案。请参阅自定义垂直分区中的示例。

参数：

mapper – 可选的映射类或对应的Mapper实例。绑定可以首先通过查阅与此Session关联的“binds”映射，其次通过查阅Mapper映射到的Table关联的MetaData进行绑定。
clause – 一个ClauseElement（例如select()，text()等）。如果未提供mapper参数或无法生成绑定，则将搜索给定的表达式构造以查找绑定元素，通常是与绑定的MetaData关联的Table。

另请参阅

分区策略（例如每个 Session 的多个数据库后端）

Session.binds

Session.bind_mapper()

Session.bind_table()

class sqlalchemy.ext.horizontal_shard.set_shard_id

语句的加载器选项，可将特定的 shard id 应用于主查询以及用于额外的关系和列加载器。

可以使用任何可执行语句的Executable.options()方法应用set_shard_id选项：

stmt = (
    select(MyObject).
    where(MyObject.name == 'some name').
    options(set_shard_id("shard1"))
)

在上述情况下，当调用语句时，主查询以及所有关系和列加载策略，包括诸如selectinload()、延迟列加载器defer()和惰性关系加载器lazyload()等都将限制为“shard1”分片标识符。

这样，set_shard_id选项的范围比在Session.execute.bind_arguments字典中使用“shard_id”参数要广泛得多。

2.0.0 版本中的新功能。

成员

init(), propagate_to_loaders

类签名

类sqlalchemy.ext.horizontal_shard.set_shard_id（sqlalchemy.orm.ORMOption）

method __init__(shard_id: str, propagate_to_loaders: bool = True)

构造一个set_shard_id选项。

参数：

shard_id – 分片标识符
propagate_to_loaders – 如果保持默认值True，则shard选项将对懒加载器（如lazyload()和defer()）生效；如果为False，则该选项将不会传播到已加载的对象。请注意，defer()无论如何都会限制到父行的shard_id，因此该参数仅对lazyload()策略的行为产生净效果。

attribute propagate_to_loaders

如果为True，表示此选项应传递到关系懒加载器以及属性加载/刷新操作中发生的“次要”SELECT 语句。

class sqlalchemy.ext.horizontal_shard.ShardedQuery

与ShardedSession一起使用的查询类。

遗留特性

ShardedQuery是遗留Query类的子类。ShardedSession现在通过ShardedSession.execute()方法支持 2.0 风格的执行。

成员

set_shard()

类签名

类 sqlalchemy.ext.horizontal_shard.ShardedQuery（sqlalchemy.orm.Query）

method set_shard(shard_id: str) → Self

返回一个新的查询，限定在单个分片 ID。

返回的查询的所有后续操作都将针对单个分片进行，而不考虑其他状态。

shard_id 可以传递给 Session.execute() 的 bind_arguments 字典，用于执行 2.0 风格的操作：

results = session.execute(
    stmt,
    bind_arguments={"shard_id": "my_shard"}
)

混合属性

原文：docs.sqlalchemy.org/en/20/orm/extensions/hybrid.html

定义在 ORM 映射类上具有“混合”行为的属性。

“混合”意味着属性在类级别和实例级别具有不同的行为。

hybrid 扩展提供了一种特殊形式的方法装饰器，并且对 SQLAlchemy 的其余部分具有最小的依赖性。它的基本操作理论可以与任何基于描述符的表达式系统一起使用。

考虑一个映射 Interval，表示整数 start 和 end 值。我们可以在映射类上定义更高级别的函数，这些函数在类级别生成 SQL 表达式，并在实例级别进行 Python 表达式评估。下面，每个使用 hybrid_method 或 hybrid_property 装饰的函数可能会接收 self 作为类的实例，或者直接接收类，具体取决于上下文：

from __future__ import annotations

from sqlalchemy.ext.hybrid import hybrid_method
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class Interval(Base):
    __tablename__ = 'interval'

    id: Mapped[int] = mapped_column(primary_key=True)
    start: Mapped[int]
    end: Mapped[int]

    def __init__(self, start: int, end: int):
        self.start = start
        self.end = end

    @hybrid_property
    def length(self) -> int:
        return self.end - self.start

    @hybrid_method
    def contains(self, point: int) -> bool:
        return (self.start <= point) & (point <= self.end)

    @hybrid_method
    def intersects(self, other: Interval) -> bool:
        return self.contains(other.start) | self.contains(other.end)

上面，length 属性返回 end 和 start 属性之间的差异。对于 Interval 的实例，这个减法在 Python 中发生，使用正常的 Python 描述符机制：

>>> i1 = Interval(5, 10)
>>> i1.length
5

处理 Interval 类本身时，hybrid_property 描述符将函数体评估为给定 Interval 类作为参数，当使用 SQLAlchemy 表达式机制评估时，将返回新的 SQL 表达式：

>>> from sqlalchemy import select
>>> print(select(Interval.length))
SELECT  interval."end"  -  interval.start  AS  length
FROM  interval
>>> print(select(Interval).filter(Interval.length > 10))
SELECT  interval.id,  interval.start,  interval."end"
FROM  interval
WHERE  interval."end"  -  interval.start  >  :param_1

过滤方法如 Select.filter_by() 也支持混合属性：

>>> print(select(Interval).filter_by(length=5))
SELECT  interval.id,  interval.start,  interval."end"
FROM  interval
WHERE  interval."end"  -  interval.start  =  :param_1

Interval 类示例还说明了两种方法，contains() 和 intersects()，使用 hybrid_method 装饰。这个装饰器将相同的思想应用于方法，就像 hybrid_property 将其应用于属性一样。这些方法返回布尔值，并利用 Python 的 | 和 & 位运算符产生等效的实例级和 SQL 表达式级布尔行为：

>>> i1.contains(6)
True
>>> i1.contains(15)
False
>>> i1.intersects(Interval(7, 18))
True
>>> i1.intersects(Interval(25, 29))
False

>>> print(select(Interval).filter(Interval.contains(15)))
SELECT  interval.id,  interval.start,  interval."end"
FROM  interval
WHERE  interval.start  <=  :start_1  AND  interval."end"  >  :end_1
>>> ia = aliased(Interval)
>>> print(select(Interval, ia).filter(Interval.intersects(ia)))
SELECT  interval.id,  interval.start,
interval."end",  interval_1.id  AS  interval_1_id,
interval_1.start  AS  interval_1_start,  interval_1."end"  AS  interval_1_end
FROM  interval,  interval  AS  interval_1
WHERE  interval.start  <=  interval_1.start
  AND  interval."end"  >  interval_1.start
  OR  interval.start  <=  interval_1."end"
  AND  interval."end"  >  interval_1."end"

定义与属性行为不同的表达行为

在前一节中，我们在 Interval.contains 和 Interval.intersects 方法中使用 & 和 | 按位运算符是幸运的，考虑到我们的函数操作两个布尔值以返回一个新值。在许多情况下，Python 函数的构建和 SQLAlchemy SQL 表达式有足够的差异，因此应该定义两个独立的 Python 表达式。hybrid 装饰器为此目的定义了一个 修饰符 hybrid_property.expression()。作为示例，我们将定义区间的半径，这需要使用绝对值函数：

from sqlalchemy import ColumnElement
from sqlalchemy import Float
from sqlalchemy import func
from sqlalchemy import type_coerce

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.expression
    @classmethod
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

在上述示例中，首先分配给名称 Interval.radius 的 hybrid_property 在后续使用 Interval._radius_expression 方法进行修改，使用装饰器 @radius.inplace.expression，将两个修饰符 hybrid_property.inplace 和 hybrid_property.expression 连接在一起。使用 hybrid_property.inplace 指示 hybrid_property.expression() 修饰符应在原地突变现有的混合对象 Interval.radius，而不是创建一个新对象。有关此修饰符及其基本原理的注释将在下一节使用 inplace 创建符合 pep-484 的混合属性中讨论。使用 @classmethod 是可选的，严格来说是为了给类型提示工具一个提示，即这种情况下 cls 应该是 Interval 类，而不是 Interval 的实例。

注意

hybrid_property.inplace 以及使用 @classmethod 进行正确类型支持的功能在 SQLAlchemy 2.0.4 中可用，之前的版本不支持。

Interval.radius 现在包含一个表达式元素，当在类级别访问 Interval.radius 时，会返回 SQL 函数 ABS()：

>>> from sqlalchemy import select
>>> print(select(Interval).filter(Interval.radius > 5))
SELECT  interval.id,  interval.start,  interval."end"
FROM  interval
WHERE  abs(interval."end"  -  interval.start)  /  :abs_1  >  :param_1 
```  ## 使用 `inplace` 创建符合 pep-484 的混合属性

在前一节中，说明了一个`hybrid_property`装饰器，其中包含两个独立的方法级函数被装饰，都用于生成一个称为`Interval.radius`的单个对象属性。实际上，我们可以使用几种不同的修饰符来修饰`hybrid_property`，包括`hybrid_property.expression()`、`hybrid_property.setter()`和`hybrid_property.update_expression()`。

SQLAlchemy 的`hybrid_property`装饰器意味着可以以与 Python 内置的`@property`装饰器相同的方式添加这些方法，其中惯用的用法是继续重定义属性，每次都使用**相同的属性名称**，就像下面的示例中演示的那样，说明了使用`hybrid_property.setter()`和`hybrid_property.expression()`来描述`Interval.radius`的用法：

```py
# correct use, however is not accepted by pep-484 tooling

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self):
        return abs(self.length) / 2

    @radius.setter
    def radius(self, value):
        self.length = value * 2

    @radius.expression
    def radius(cls):
        return type_coerce(func.abs(cls.length) / 2, Float)

如上所述，有三个Interval.radius方法，但由于每个都被hybrid_property装饰器和@radius名称本身装饰，因此最终效果是Interval.radius是一个具有三个不同功能的单个属性。这种使用方式取自于Python 文档中对@property 的使用。值得注意的是，@property以及hybrid_property的工作方式，每次都会复制描述符。也就是说，每次调用@radius.expression、@radius.setter等都会完全创建一个新对象。这允许在子类中重新定义属性而无需问题（请参阅本节稍后的在子类中重用混合属性的使用方式）。

然而，上述方法与 mypy 和 pyright 等类型工具不兼容。 Python 自己的@property装饰器之所以没有此限制，只是因为这些工具硬编码了@property 的行为，这意味着此语法不符合PEP 484的要求。

为了产生合理的语法，同时保持类型兼容性，hybrid_property.inplace装饰器允许使用不同的方法名重复使用相同的装饰器，同时仍然在一个名称下生成一个装饰器：

# correct use which is also accepted by pep-484 tooling

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.setter
    def _radius_setter(self, value: float) -> None:
        # for example only
        self.length = value * 2

    @radius.inplace.expression
    @classmethod
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

使用hybrid_property.inplace进一步限定了应该不制作新副本的装饰器的使用，从而保持了Interval.radius名称，同时允许其他方法Interval._radius_setter和Interval._radius_expression命名不同。

2.0.4 版中的新功能：添加了hybrid_property.inplace，以允许更少冗长的构造复合hybrid_property对象，同时无需使用重复的方法名称。此外，允许在hybrid_property.expression、hybrid_property.update_expression和hybrid_property.comparator内使用@classmethod，以允许类型工具将cls识别为类而不是方法签名中的实例。

定义设置器

hybrid_property.setter()修饰符允许构造自定义的设置器方法，可以修改对象上的值：

class Interval(Base):
    # ...

    @hybrid_property
    def length(self) -> int:
        return self.end - self.start

    @length.inplace.setter
    def _length_setter(self, value: int) -> None:
        self.end = self.start + value

现在，在设置时调用length(self, value)方法：

>>> i1 = Interval(5, 10)
>>> i1.length
5
>>> i1.length = 12
>>> i1.end
17

允许批量 ORM 更新

一个混合可以为启用 ORM 的更新定义自定义的“UPDATE”处理程序，从而允许混合用于更新的 SET 子句中。

通常，在使用带有update()的混合时，SQL 表达式被用作作为 SET 目标的列。如果我们的Interval类有一个混合start_point，它链接到Interval.start，这可以直接替换：

from sqlalchemy import update
stmt = update(Interval).values({Interval.start_point: 10})

但是，当使用类似Interval.length的复合混合类型时，此混合类型表示不止一个列。我们可以设置一个处理程序，该处理程序将适应传递给 VALUES 表达式的值，这可能会影响到这一点，使用hybrid_property.update_expression()装饰器。一个类似于我们的设置器的处理程序将是：

from typing import List, Tuple, Any

class Interval(Base):
    # ...

    @hybrid_property
    def length(self) -> int:
        return self.end - self.start

    @length.inplace.setter
    def _length_setter(self, value: int) -> None:
        self.end = self.start + value

    @length.inplace.update_expression
    def _length_update_expression(cls, value: Any) -> List[Tuple[Any, Any]]:
        return [
            (cls.end, cls.start + value)
        ]

以上，如果我们在 UPDATE 表达式中使用Interval.length，我们将得到一个混合 SET 表达式：

>>> from sqlalchemy import update
>>> print(update(Interval).values({Interval.length: 25}))
UPDATE  interval  SET  "end"=(interval.start  +  :start_1)

这个 SET 表达式会被 ORM 自动处理。

另请参阅

ORM-启用的 INSERT、UPDATE 和 DELETE 语句 - 包括 ORM 启用的 UPDATE 语句的背景信息

处理关系

创建与基于列的数据不同的混合对象时，本质上没有区别。对于不同的表达式的需求往往更大。我们将展示的两种变体是“连接依赖”混合和“相关子查询”混合。

连接依赖关系混合

考虑以下将 User 与 SavingsAccount 关联的声明性映射：

from __future__ import annotations

from decimal import Decimal
from typing import cast
from typing import List
from typing import Optional

from sqlalchemy import ForeignKey
from sqlalchemy import Numeric
from sqlalchemy import String
from sqlalchemy import SQLColumnExpression
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class SavingsAccount(Base):
    __tablename__ = 'account'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    balance: Mapped[Decimal] = mapped_column(Numeric(15, 5))

    owner: Mapped[User] = relationship(back_populates="accounts")

class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    accounts: Mapped[List[SavingsAccount]] = relationship(
        back_populates="owner", lazy="selectin"
    )

    @hybrid_property
    def balance(self) -> Optional[Decimal]:
        if self.accounts:
            return self.accounts[0].balance
        else:
            return None

    @balance.inplace.setter
    def _balance_setter(self, value: Optional[Decimal]) -> None:
        assert value is not None

        if not self.accounts:
            account = SavingsAccount(owner=self)
        else:
            account = self.accounts[0]
        account.balance = value

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Optional[Decimal]]:
        return cast("SQLColumnExpression[Optional[Decimal]]", SavingsAccount.balance)

上面的混合属性 balance 与此用户的账户列表中的第一个 SavingsAccount 条目一起工作。在 Python 中的 getter/setter 方法可以将 accounts 视为可在 self 上使用的 Python 列表。

提示

在上面的例子中，User.balance 的 getter 方法访问了 self.accounts 集合，通常会通过配置在 User.balance 的 relationship() 上的 selectinload() 加载策略来加载。当在 relationship() 上没有另外指定时，默认的加载策略是 lazyload()，它会按需发出 SQL。在使用 asyncio 时，像 lazyload() 这样的按需加载器不受支持，因此在使用 asyncio 时，应确保 self.accounts 集合对这个混合访问器是可访问的。

在表达式级别，预期 User 类将在适当的上下文中使用，以便存在与 SavingsAccount 的适当连接：

>>> from sqlalchemy import select
>>> print(select(User, User.balance).
...       join(User.accounts).filter(User.balance > 5000))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  >  :balance_1

但需要注意的是，尽管实例级别的访问器需要担心 self.accounts 是否存在，但在 SQL 表达式级别，这个问题表现得不同，我们基本上会使用外连接：

>>> from sqlalchemy import select
>>> from sqlalchemy import or_
>>> print (select(User, User.balance).outerjoin(User.accounts).
...         filter(or_(User.balance < 5000, User.balance == None)))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  LEFT  OUTER  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  <  :balance_1  OR  account.balance  IS  NULL

构建自定义比较器

混合属性还包括一个辅助程序，允许构建自定义比较器。比较器对象允许单独定制每个 SQLAlchemy 表达式操作符的行为。在创建在 SQL 方面具有某些高度特殊行为的自定义类型时很有用。

注意

本节中引入的hybrid_property.comparator()装饰器替换了hybrid_property.expression()装饰器的使用。它们不能一起使用。

下面的示例类允许在名为word_insensitive的属性上进行不区分大小写的比较：

from __future__ import annotations

from typing import Any

from sqlalchemy import ColumnElement
from sqlalchemy import func
from sqlalchemy.ext.hybrid import Comparator
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class CaseInsensitiveComparator(Comparator[str]):
    def __eq__(self, other: Any) -> ColumnElement[bool]:  # type: ignore[override]  # noqa: E501
        return func.lower(self.__clause_element__()) == func.lower(other)

class SearchWord(Base):
    __tablename__ = 'searchword'

    id: Mapped[int] = mapped_column(primary_key=True)
    word: Mapped[str]

    @hybrid_property
    def word_insensitive(self) -> str:
        return self.word.lower()

    @word_insensitive.inplace.comparator
    @classmethod
    def _word_insensitive_comparator(cls) -> CaseInsensitiveComparator:
        return CaseInsensitiveComparator(cls.word)

在上述情况下，针对word_insensitive的 SQL 表达式将对两侧应用LOWER() SQL 函数：

>>> from sqlalchemy import select
>>> print(select(SearchWord).filter_by(word_insensitive="Trucks"))
SELECT  searchword.id,  searchword.word
FROM  searchword
WHERE  lower(searchword.word)  =  lower(:lower_1)

上述的CaseInsensitiveComparator实现了ColumnOperators接口的部分内容。可以使用Operators.operate()对所有比较操作（即eq、lt、gt等）应用“强制转换”操作，如转换为小写：

class CaseInsensitiveComparator(Comparator):
    def operate(self, op, other, **kwargs):
        return op(
            func.lower(self.__clause_element__()),
            func.lower(other),
            **kwargs,
        )
```  ## 在子类之间重用混合属性

可以从超类中引用混合体，以允许修改方法，如`hybrid_property.getter()`，`hybrid_property.setter()`，以便在子类中重新定义这些方法。这类似于标准 Python 的`@property`对象的工作原理：

```py
class FirstNameOnly(Base):
    # ...

    first_name: Mapped[str]

    @hybrid_property
    def name(self) -> str:
        return self.first_name

    @name.inplace.setter
    def _name_setter(self, value: str) -> None:
        self.first_name = value

class FirstNameLastName(FirstNameOnly):
    # ...

    last_name: Mapped[str]

    # 'inplace' is not used here; calling getter creates a copy
    # of FirstNameOnly.name that is local to FirstNameLastName
    @FirstNameOnly.name.getter
    def name(self) -> str:
        return self.first_name + ' ' + self.last_name

    @name.inplace.setter
    def _name_setter(self, value: str) -> None:
        self.first_name, self.last_name = value.split(' ', 1)

在上述情况下，FirstNameLastName类引用了从FirstNameOnly.name到子类的混合体，以重新利用其 getter 和 setter。

当仅在首次引用超类时覆盖hybrid_property.expression()和hybrid_property.comparator()作为类级别的第一个引用时，这些名称会与返回类级别的QueryableAttribute对象上的同名访问器发生冲突。要在直接引用父类描述符时覆盖这些方法，请添加特殊限定词hybrid_property.overrides，该限定词将仪表化的属性引用回混合对象：

class FirstNameLastName(FirstNameOnly):
    # ...

    last_name: Mapped[str]

    @FirstNameOnly.name.overrides.expression
    @classmethod
    def name(cls):
        return func.concat(cls.first_name, ' ', cls.last_name)

混合值对象

请注意，在我们之前的例子中，如果我们将SearchWord实例的word_insensitive属性与普通的 Python 字符串进行比较，普通的 Python 字符串不会被强制转换为小写 - 我们构建的CaseInsensitiveComparator，由@word_insensitive.comparator返回，仅适用于 SQL 端。

自定义比较器的更全面形式是构建一个混合值对象。这种技术将目标值或表达式应用于一个值对象，然后由访问器在所有情况下返回。值对象允许控制对值的所有操作，以及如何处理比较的值，无论是在 SQL 表达式方面还是在 Python 值方面。用新的CaseInsensitiveWord类替换以前的CaseInsensitiveComparator类：

class CaseInsensitiveWord(Comparator):
    "Hybrid value representing a lower case representation of a word."

    def __init__(self, word):
        if isinstance(word, basestring):
            self.word = word.lower()
        elif isinstance(word, CaseInsensitiveWord):
            self.word = word.word
        else:
            self.word = func.lower(word)

    def operate(self, op, other, **kwargs):
        if not isinstance(other, CaseInsensitiveWord):
            other = CaseInsensitiveWord(other)
        return op(self.word, other.word, **kwargs)

    def __clause_element__(self):
        return self.word

    def __str__(self):
        return self.word

    key = 'word'
    "Label to apply to Query tuple results"

在上面的例子中，CaseInsensitiveWord对象表示self.word，它可能是一个 SQL 函数，也可能是一个 Python 本机对象。通过重写operate()和__clause_element__()以使用self.word，所有比较操作将针对“转换”形式的word进行，无论是在 SQL 端还是在 Python 端。我们的SearchWord类现在可以无条件地从单个混合调用中交付CaseInsensitiveWord对象：

class SearchWord(Base):
    __tablename__ = 'searchword'
    id: Mapped[int] = mapped_column(primary_key=True)
    word: Mapped[str]

    @hybrid_property
    def word_insensitive(self) -> CaseInsensitiveWord:
        return CaseInsensitiveWord(self.word)

word_insensitive属性现在在所有情况下都具有不区分大小写的比较行为，包括 SQL 表达式与 Python 表达式（请注意，此处 Python 值在 Python 端转换为小写）：

>>> print(select(SearchWord).filter_by(word_insensitive="Trucks"))
SELECT  searchword.id  AS  searchword_id,  searchword.word  AS  searchword_word
FROM  searchword
WHERE  lower(searchword.word)  =  :lower_1

SQL 表达式与 SQL 表达式：

>>> from sqlalchemy.orm import aliased
>>> sw1 = aliased(SearchWord)
>>> sw2 = aliased(SearchWord)
>>> print(
...     select(sw1.word_insensitive, sw2.word_insensitive).filter(
...         sw1.word_insensitive > sw2.word_insensitive
...     )
... )
SELECT  lower(searchword_1.word)  AS  lower_1,
lower(searchword_2.word)  AS  lower_2
FROM  searchword  AS  searchword_1,  searchword  AS  searchword_2
WHERE  lower(searchword_1.word)  >  lower(searchword_2.word)

仅 Python 表达式：

>>> ws1 = SearchWord(word="SomeWord")
>>> ws1.word_insensitive == "sOmEwOrD"
True
>>> ws1.word_insensitive == "XOmEwOrX"
False
>>> print(ws1.word_insensitive)
someword

混合值模式非常适用于可能具有多种表示形式的任何类型的值，例如时间戳、时间差、测量单位、货币和加密密码。

另请参阅

混合和值不可知类型 - 在 techspot.zzzeek.org 博客上

值不可知类型，第二部分 - 在 techspot.zzzeek.org 博客上

API 参考

对象名称	描述
比较器	一个帮助类，允许轻松构建用于混合使用的自定义`PropComparator`类。
hybrid_method	一个装饰器，允许定义具有实例级和类级行为的 Python 对象方法。
hybrid_property	一个装饰器，允许定义具有实例级和类级行为的 Python 描述符。
混合扩展类型	一个枚举类型。

class sqlalchemy.ext.hybrid.hybrid_method

一个装饰器，允许定义具有实例级和类级行为的 Python 对象方法。

成员

init(), expression(), extension_type, inplace, is_attribute

类签名

类sqlalchemy.ext.hybrid.hybrid_method（sqlalchemy.orm.base.InspectionAttrInfo, typing.Generic)

method __init__(func: Callable[[Concatenate[Any, _P]], _R], expr: Callable[[Concatenate[Any, _P]], SQLCoreOperations[_R]] | None = None)

创建一个新的hybrid_method。

通常使用装饰器：

from sqlalchemy.ext.hybrid import hybrid_method

class SomeClass:
    @hybrid_method
    def value(self, x, y):
        return self._value + x + y

    @value.expression
    @classmethod
    def value(cls, x, y):
        return func.some_function(cls._value, x, y)

method expression(expr: Callable[[Concatenate[Any, _P]], SQLCoreOperations[_R]]) → hybrid_method[_P, _R]

提供一个修改装饰器，定义一个生成 SQL 表达式的方法。

attribute extension_type: InspectionAttrExtensionType = 'HYBRID_METHOD'

扩展类型，如果有的话。默认为NotExtension.NOT_EXTENSION

另请参阅

HybridExtensionType

AssociationProxyExtensionType

attribute inplace

返回此hybrid_method的原地变异器。

当调用hybrid_method.expression()装饰器时，hybrid_method类已经执行“原地”变异，因此此属性返回 Self。

版本 2.0.4 中的新功能。

另请参阅

使用 inplace 创建符合 pep-484 的混合属性

attribute is_attribute = True

如果这个对象是一个 Python 描述符，则为 True。

这可以指代许多类型之一。通常是一个处理属性事件的QueryableAttribute，代表一个MapperProperty。但也可以是一个扩展类型，如AssociationProxy或hybrid_property。InspectionAttr.extension_type将指代一个标识特定子类型的常量。

另请参阅

Mapper.all_orm_descriptors

class sqlalchemy.ext.hybrid.hybrid_property

一个允许定义具有实例级和类级行为的 Python 描述符的装饰器。

成员

init(), comparator(), deleter(), expression(), extension_type, getter(), inplace, is_attribute, overrides, setter(), update_expression()

类签名

类sqlalchemy.ext.hybrid.hybrid_property（sqlalchemy.orm.base.InspectionAttrInfo, sqlalchemy.orm.base.ORMDescriptor)

method __init__(fget: _HybridGetterType[_T], fset: _HybridSetterType[_T] | None = None, fdel: _HybridDeleterType[_T] | None = None, expr: _HybridExprCallableType[_T] | None = None, custom_comparator: Comparator[_T] | None = None, update_expr: _HybridUpdaterType[_T] | None = None)

创建一个新的hybrid_property。

通常使用修饰器：

from sqlalchemy.ext.hybrid import hybrid_property

class SomeClass:
    @hybrid_property
    def value(self):
        return self._value

    @value.setter
    def value(self, value):
        self._value = value

method comparator(comparator: _HybridComparatorCallableType[_T]) → hybrid_property[_T]

提供一个修饰器，定义一个自定义比较器生成方法。

被修饰方法的返回值应该是Comparator的一个实例。

注意

hybrid_property.comparator()修饰器替代了hybrid_property.expression()修饰器的使用。它们不能同时使用。

当在类级别调用混合属性时，这里给出的Comparator对象被包装在一个专门的QueryableAttribute中，这是 ORM 用来表示其他映射属性的对象。这样做的原因是为了在返回的结构中保留其他类级别属性，如文档字符串和对混合属性本身的引用，而不对传入的原始比较器对象进行任何修改。

注意

当引用拥有类（例如 SomeClass.some_hybrid）的混合属性时，会返回一个QueryableAttribute的实例，表示此混合对象的表达式或比较器对象。但是，该对象本身具有名为 expression 和 comparator 的访问器；因此，在尝试在子类上覆盖这些装饰器时，可能需要首先使用 hybrid_property.overrides 修饰符进行限定。有关详细信息，请参阅该修饰符。

method deleter(fdel: _HybridDeleterType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个删除方法。

method expression(expr: _HybridExprCallableType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个生成 SQL 表达式的方法。

当在类级别调用混合时，此处给出的 SQL 表达式将包装在一个专门的 QueryableAttribute 内，这是 ORM 用于表示其他映射属性的相同类型的对象。这样做的原因是为了在返回的结构中维护其他类级别属性，例如文档字符串和混合本身的引用，而不对传入的原始 SQL 表达式进行任何修改。

注意

当引用拥有类（例如 SomeClass.some_hybrid）的混合属性时，会返回一个QueryableAttribute的实例，表示表达式或比较器对象以及此混合对象。但是，该对象本身具有名为 expression 和 comparator 的访问器；因此，在尝试在子类上覆盖这些装饰器时，可能需要首先使用 hybrid_property.overrides 修饰符进行限定。有关详细信息，请参阅该修饰符。

请参见

定义与属性行为不同的表达行为

attribute extension_type: InspectionAttrExtensionType = 'HYBRID_PROPERTY'

扩展类型，如果有的话。默认为 NotExtension.NOT_EXTENSION

请参见

HybridExtensionType

AssociationProxyExtensionType

method getter(fget: _HybridGetterType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个获取器方法。

自 1.2 版开始新添加。

attribute inplace

返回此 hybrid_property 的原地变异器。

这是为了允许对混合进行就地变异，允许重用特定名称的第一个混合方法以添加更多方法，而不必将这些方法命名为相同的名称，例如：

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.setter
    def _radius_setter(self, value: float) -> None:
        self.length = value * 2

    @radius.inplace.expression
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

自 2.0.4 版开始新添加。

请参见

使用 inplace 创建符合 pep-484 的混合属性

attribute is_attribute = True

如果此对象是 Python 的描述符，则为 True。

这可能是多种类型之一。通常是一个QueryableAttribute，它代表一个MapperProperty上的属性事件。但也可以是诸如AssociationProxy或hybrid_property之类的扩展类型。InspectionAttr.extension_type 将引用一个常量，以识别特定的子类型。

另请参阅

Mapper.all_orm_descriptors

attribute overrides

重写现有属性的方法的前缀。

hybrid_property.overrides 访问器只是返回这个混合对象，当在父类的类级别从父类调用时，它将取消引用通常在此级别返回的“instrumented attribute”，并允许修改装饰器，例如 hybrid_property.expression() 和 hybrid_property.comparator() 被使用而不与通常存在于QueryableAttribute上的同名属性冲突：

class SuperClass:
    # ...

    @hybrid_property
    def foobar(self):
        return self._foobar

class SubClass(SuperClass):
    # ...

    @SuperClass.foobar.overrides.expression
    def foobar(cls):
        return func.subfoobar(self._foobar)

1.2 版的新功能。

另请参阅

在子类中重用混合属性

method setter(fset: _HybridSetterType[_T]) → hybrid_property[_T]

提供一个定义 setter 方法的修改装饰器。

method update_expression(meth: _HybridUpdaterType[_T]) → hybrid_property[_T]

提供一个定义产生 UPDATE 元组的修改装饰器方法。

该方法接受一个值，该值将被渲染到 UPDATE 语句的 SET 子句中。然后，该方法应将此值处理为适合最终 SET 子句的单个列表达式，并将它们作为 2 元组的序列返回。每个元组包含一个列表达式作为键和要渲染的值。

例如：

class Person(Base):
    # ...

    first_name = Column(String)
    last_name = Column(String)

    @hybrid_property
    def fullname(self):
        return first_name + " " + last_name

    @fullname.update_expression
    def fullname(cls, value):
        fname, lname = value.split(" ", 1)
        return [
            (cls.first_name, fname),
            (cls.last_name, lname)
        ]

1.2 版的新功能。

class sqlalchemy.ext.hybrid.Comparator

一个辅助类，允许轻松构建自定义PropComparator类以与混合一起使用。

类签名

类sqlalchemy.ext.hybrid.Comparator (sqlalchemy.orm.PropComparator)

class sqlalchemy.ext.hybrid.HybridExtensionType

一个枚举。

成员

HYBRID_METHOD, HYBRID_PROPERTY

类签名

类sqlalchemy.ext.hybrid.HybridExtensionType (sqlalchemy.orm.base.InspectionAttrExtensionType)

attribute HYBRID_METHOD = 'HYBRID_METHOD'

表示一个InspectionAttr的符号，其类型为hybrid_method。

被赋予InspectionAttr.extension_type属性。

另请参阅

Mapper.all_orm_attributes

attribute HYBRID_PROPERTY = 'HYBRID_PROPERTY'

表示一个InspectionAttr的符号

类型为hybrid_method。

被赋予InspectionAttr.extension_type属性。

另请参阅

Mapper.all_orm_attributes

定义与属性行为不同的表达行为

在前一节中，我们在Interval.contains和Interval.intersects方法中使用&和|位运算符是幸运的，考虑到我们的函数操作两个布尔值以返回一个新值。在许多情况下，一个在 Python 函数和 SQLAlchemy SQL 表达式之间有足够差异的情况下，应该定义两个单独的 Python 表达式。hybrid装饰器为此目的定义了一个修饰符hybrid_property.expression()。作为示例，我们将定义间隔的半径，这需要使用绝对值函数：

from sqlalchemy import ColumnElement
from sqlalchemy import Float
from sqlalchemy import func
from sqlalchemy import type_coerce

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.expression
    @classmethod
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

在上面的示例中，首先将 hybrid_property 分配给名称 Interval.radius，然后通过后续调用名为 Interval._radius_expression 的方法，使用装饰器 @radius.inplace.expression 对其进行修改，该装饰器链接了两个修饰符 hybrid_property.inplace 和 hybrid_property.expression。使用 hybrid_property.inplace 表示 hybrid_property.expression() 修饰符应在原地改变 Interval.radius 中的现有混合对象，而不创建新对象。关于此修饰符及其基本原理的说明在下一节中讨论使用 inplace 创建符合 pep-484 的混合属性。使用 @classmethod 是可选的，并且严格来说是为了给出类型提示，以表明在这种情况下，cls 预期是 Interval 类，而不是 Interval 的实例。

注意

hybrid_property.inplace 以及使用 @classmethod 以获得正确的类型支持在 SQLAlchemy 2.0.4 中可用，并且在早期版本中不起作用。

现在，由于 Interval.radius 现在包含表达式元素，因此在访问类级别的 Interval.radius 时会返回 SQL 函数 ABS()：

>>> from sqlalchemy import select
>>> print(select(Interval).filter(Interval.radius > 5))
SELECT  interval.id,  interval.start,  interval."end"
FROM  interval
WHERE  abs(interval."end"  -  interval.start)  /  :abs_1  >  :param_1

使用 `inplace` 创建符合 pep-484 的混合属性

在前一节中，演示了一个 hybrid_property 装饰器，其中包括两个单独的方法级函数被装饰，都用于生成一个名为 Interval.radius 的单个对象属性。实际上，我们可以使用几种不同的修饰符来使用 hybrid_property，包括 hybrid_property.expression()、hybrid_property.setter() 和 hybrid_property.update_expression()。

SQLAlchemy 的 hybrid_property 装饰器打算通过与 Python 内置的 @property 装饰器相同的方式添加这些方法，其中惯用法是重复重新定义属性，每次使用相同的属性名称，如下面的示例所示，演示了使用 hybrid_property.setter() 和 hybrid_property.expression() 为 Interval.radius 描述符的用法：

# correct use, however is not accepted by pep-484 tooling

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self):
        return abs(self.length) / 2

    @radius.setter
    def radius(self, value):
        self.length = value * 2

    @radius.expression
    def radius(cls):
        return type_coerce(func.abs(cls.length) / 2, Float)

如上，有三个 Interval.radius 方法，但是由于每个都被 hybrid_property 装饰器和 @radius 名称本身装饰，最终的效果是 Interval.radius 是一个包含三个不同函数的单个属性。这种用法风格来自于 Python 的@property 的文档用法。需要注意的是，无论是 @property 还是 hybrid_property 的工作方式，每次都会 复制描述符。也就是说，每次调用 @radius.expression、@radius.setter 等都会完全创建一个新的对象。这使得属性在子类中重新定义时不会出现问题（请参阅本节稍后的在子类之间重用混合属性来了解如何使用）。

然而，上述方法不兼容于诸如 mypy 和 pyright 等类型工具。Python 自己的 @property 装饰器之所以没有这个限制，仅仅是因为这些工具硬编码了@property 的行为，这意味着这种语法在 SQLAlchemy 下不符合 PEP 484 的规范。

为了在保持打字兼容的同时产生合理的语法，hybrid_property.inplace 装饰器允许同一装饰器以不同的方法名称被重复使用，同时仍然产生一个单一的装饰器在一个名称下：

# correct use which is also accepted by pep-484 tooling

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.setter
    def _radius_setter(self, value: float) -> None:
        # for example only
        self.length = value * 2

    @radius.inplace.expression
    @classmethod
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

使用 hybrid_property.inplace 进一步限定了装饰器的使用，不应该创建一个新的副本，从而保持了 Interval.radius 名称，同时允许额外的方法 Interval._radius_setter 和 Interval._radius_expression 使用不同的名称。

版本 2.0.4 中的新功能：添加hybrid_property.inplace，以允许更简洁地构建复合hybrid_property对象，同时不必重复使用方法名称。此外，允许在hybrid_property.expression、hybrid_property.update_expression和hybrid_property.comparator中使用@classmethod，以便允许类型工具将cls识别为类而不是方法签名中的实例。

定义 Setter

hybrid_property.setter()修饰符允许构建自定义 setter 方法，可以修改对象上的值：

class Interval(Base):
    # ...

    @hybrid_property
    def length(self) -> int:
        return self.end - self.start

    @length.inplace.setter
    def _length_setter(self, value: int) -> None:
        self.end = self.start + value

现在，在设置时调用length(self, value)方法：

>>> i1 = Interval(5, 10)
>>> i1.length
5
>>> i1.length = 12
>>> i1.end
17

允许批量 ORM 更新

当使用 ORM 启用的更新时，混合类型可以为自定义的“UPDATE”处理程序定义处理程序，允许将混合类型用于更新的 SET 子句中。

通常，当使用update()与混合类型时，SQL 表达式将用作 SET 的目标列。如果我们的Interval类具有链接到Interval.start的混合类型start_point，则可以直接替换：

from sqlalchemy import update
stmt = update(Interval).values({Interval.start_point: 10})

然而，当使用像Interval.length这样的复合混合类型时，这个混合类型代表多于一个列。我们可以设置一个处理程序，该处理程序将适应传递到 VALUES 表达式中的值，这可能会影响此处理程序，使用hybrid_property.update_expression()装饰器。一个与我们的 setter 类似的处理程序将是：

from typing import List, Tuple, Any

class Interval(Base):
    # ...

    @hybrid_property
    def length(self) -> int:
        return self.end - self.start

    @length.inplace.setter
    def _length_setter(self, value: int) -> None:
        self.end = self.start + value

    @length.inplace.update_expression
    def _length_update_expression(cls, value: Any) -> List[Tuple[Any, Any]]:
        return [
            (cls.end, cls.start + value)
        ]

如果我们在 UPDATE 表达式中使用Interval.length，我们将获得一个混合类型的 SET 表达式：

>>> from sqlalchemy import update
>>> print(update(Interval).values({Interval.length: 25}))
UPDATE  interval  SET  "end"=(interval.start  +  :start_1)

此 SET 表达式将由 ORM 自动处理。

另请参阅

ORM 启用的 INSERT、UPDATE 和 DELETE 语句 - 包括 ORM 启用的 UPDATE 语句的背景信息

与关系一起工作

创建与基于列的数据相反的与相关对象一起工作的混合类型时，没有本质区别。对于不同的表达式的需求往往更大。我们将说明的两种变体是“join-dependent”混合类型和“correlated subquery”混合类型。

Join-Dependent Relationship Hybrid

考虑以下将User与SavingsAccount相关联的声明性映射：

from __future__ import annotations

from decimal import Decimal
from typing import cast
from typing import List
from typing import Optional

from sqlalchemy import ForeignKey
from sqlalchemy import Numeric
from sqlalchemy import String
from sqlalchemy import SQLColumnExpression
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class SavingsAccount(Base):
    __tablename__ = 'account'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    balance: Mapped[Decimal] = mapped_column(Numeric(15, 5))

    owner: Mapped[User] = relationship(back_populates="accounts")

class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    accounts: Mapped[List[SavingsAccount]] = relationship(
        back_populates="owner", lazy="selectin"
    )

    @hybrid_property
    def balance(self) -> Optional[Decimal]:
        if self.accounts:
            return self.accounts[0].balance
        else:
            return None

    @balance.inplace.setter
    def _balance_setter(self, value: Optional[Decimal]) -> None:
        assert value is not None

        if not self.accounts:
            account = SavingsAccount(owner=self)
        else:
            account = self.accounts[0]
        account.balance = value

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Optional[Decimal]]:
        return cast("SQLColumnExpression[Optional[Decimal]]", SavingsAccount.balance)

上述混合属性balance与此用户的账户列表中的第一个SavingsAccount条目配合使用。在 Python 中，getter/setter 方法可以将accounts视为self上可用的 Python 列表。

提示

上述示例中的User.balance getter 访问self.acccounts集合，通常会通过配置在User.balance relationship()上的selectinload()加载策略加载。当未在relationship()上另行说明时，默认加载策略是lazyload()，它会按需发出 SQL。在使用 asyncio 时，不支持按需加载程序，因此在使用 asyncio 时，应确保self.accounts集合对此混合访问器可访问。

在表达式级别，预期User类将在适当的上下文中使用，以便存在适当的连接到SavingsAccount：

>>> from sqlalchemy import select
>>> print(select(User, User.balance).
...       join(User.accounts).filter(User.balance > 5000))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  >  :balance_1

但请注意，尽管实例级访问器需要担心self.accounts是否存在，但在 SQL 表达式级别，这个问题表现得不同，基本上我们会使用外连接：

>>> from sqlalchemy import select
>>> from sqlalchemy import or_
>>> print (select(User, User.balance).outerjoin(User.accounts).
...         filter(or_(User.balance < 5000, User.balance == None)))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  LEFT  OUTER  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  <  :balance_1  OR  account.balance  IS  NULL

连接依赖关系混合

考虑以下声明性映射，将User与SavingsAccount相关联：

from __future__ import annotations

from decimal import Decimal
from typing import cast
from typing import List
from typing import Optional

from sqlalchemy import ForeignKey
from sqlalchemy import Numeric
from sqlalchemy import String
from sqlalchemy import SQLColumnExpression
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class SavingsAccount(Base):
    __tablename__ = 'account'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    balance: Mapped[Decimal] = mapped_column(Numeric(15, 5))

    owner: Mapped[User] = relationship(back_populates="accounts")

class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    accounts: Mapped[List[SavingsAccount]] = relationship(
        back_populates="owner", lazy="selectin"
    )

    @hybrid_property
    def balance(self) -> Optional[Decimal]:
        if self.accounts:
            return self.accounts[0].balance
        else:
            return None

    @balance.inplace.setter
    def _balance_setter(self, value: Optional[Decimal]) -> None:
        assert value is not None

        if not self.accounts:
            account = SavingsAccount(owner=self)
        else:
            account = self.accounts[0]
        account.balance = value

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Optional[Decimal]]:
        return cast("SQLColumnExpression[Optional[Decimal]]", SavingsAccount.balance)

提示

上面示例中的User.balance getter 访问了self.acccounts集合，这通常会通过在User.balance relationship()上配置的selectinload()加载器策略进行加载。当没有在其他地方声明relationship()时，默认的加载器策略是lazyload()，它按需发出 SQL。当使用 asyncio 时，不支持按需加载器，如lazyload()，因此在使用 asyncio 时应注意确保self.accounts集合对此混合访问器是可访问的。

在表达式级别，预计User类将在适当的上下文中使用，以便存在适当的连接到SavingsAccount：

>>> from sqlalchemy import select
>>> print(select(User, User.balance).
...       join(User.accounts).filter(User.balance > 5000))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  >  :balance_1

但请注意，尽管实例级别的访问器需要担心self.accounts是否存在，但这个问题在 SQL 表达式级别上表现得不同，我们基本上会使用外连接：

>>> from sqlalchemy import select
>>> from sqlalchemy import or_
>>> print (select(User, User.balance).outerjoin(User.accounts).
...         filter(or_(User.balance < 5000, User.balance == None)))
SELECT  "user".id  AS  user_id,  "user".name  AS  user_name,
account.balance  AS  account_balance
FROM  "user"  LEFT  OUTER  JOIN  account  ON  "user".id  =  account.user_id
WHERE  account.balance  <  :balance_1  OR  account.balance  IS  NULL

关联子查询关系混合

当然，我们可以放弃依赖包含查询中的连接，而倾向于关联子查询，这可以被封装成一个单列表达式。关联子查询更具可移植性，但在 SQL 级别上通常性能较差。使用在使用 column_property 中说明的相同技术，我们可以调整我们的SavingsAccount示例来聚合所有账户的余额，并为列表达式使用关联子查询：

from __future__ import annotations

from decimal import Decimal
from typing import List

from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Numeric
from sqlalchemy import select
from sqlalchemy import SQLColumnExpression
from sqlalchemy import String
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class SavingsAccount(Base):
    __tablename__ = 'account'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    balance: Mapped[Decimal] = mapped_column(Numeric(15, 5))

    owner: Mapped[User] = relationship(back_populates="accounts")

class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    accounts: Mapped[List[SavingsAccount]] = relationship(
        back_populates="owner", lazy="selectin"
    )

    @hybrid_property
    def balance(self) -> Decimal:
        return sum((acc.balance for acc in self.accounts), start=Decimal("0"))

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Decimal]:
        return (
            select(func.sum(SavingsAccount.balance))
            .where(SavingsAccount.user_id == cls.id)
            .label("total_balance")
        )

上面的配方将为我们提供一个渲染关联 SELECT 的balance列：

>>> from sqlalchemy import select
>>> print(select(User).filter(User.balance > 400))
SELECT  "user".id,  "user".name
FROM  "user"
WHERE  (
  SELECT  sum(account.balance)  AS  sum_1  FROM  account
  WHERE  account.user_id  =  "user".id
)  >  :param_1

构建自定义比较器

混合属性还包括一个助手，允许构建自定义比较器。比较器对象允许用户单独定制每个 SQLAlchemy 表达式操作符的行为。当创建具有一些高度特殊的 SQL 端行为的自定义类型时，它们非常有用。

注意

此部分介绍的hybrid_property.comparator()装饰器替换了hybrid_property.expression()装饰器的使用。它们不能一起使用。

下面的示例类允许在名为word_insensitive的属性上进行不区分大小写的比较：

from __future__ import annotations

from typing import Any

from sqlalchemy import ColumnElement
from sqlalchemy import func
from sqlalchemy.ext.hybrid import Comparator
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class CaseInsensitiveComparator(Comparator[str]):
    def __eq__(self, other: Any) -> ColumnElement[bool]:  # type: ignore[override]  # noqa: E501
        return func.lower(self.__clause_element__()) == func.lower(other)

class SearchWord(Base):
    __tablename__ = 'searchword'

    id: Mapped[int] = mapped_column(primary_key=True)
    word: Mapped[str]

    @hybrid_property
    def word_insensitive(self) -> str:
        return self.word.lower()

    @word_insensitive.inplace.comparator
    @classmethod
    def _word_insensitive_comparator(cls) -> CaseInsensitiveComparator:
        return CaseInsensitiveComparator(cls.word)

上面，针对word_insensitive的 SQL 表达式将对双方都应用LOWER() SQL 函数：

>>> from sqlalchemy import select
>>> print(select(SearchWord).filter_by(word_insensitive="Trucks"))
SELECT  searchword.id,  searchword.word
FROM  searchword
WHERE  lower(searchword.word)  =  lower(:lower_1)

上面的CaseInsensitiveComparator实现了ColumnOperators接口的一部分。像小写化这样的“强制转换”操作可以应用于所有比较操作（即eq、lt、gt等）使用Operators.operate()：

class CaseInsensitiveComparator(Comparator):
    def operate(self, op, other, **kwargs):
        return op(
            func.lower(self.__clause_element__()),
            func.lower(other),
            **kwargs,
        )

在子类之间重用混合属性

可以从超类中引用混合，以允许修改方法，如hybrid_property.getter()、hybrid_property.setter()等，用于在子类上重新定义这些方法。这类似于标准的 Python @property对象的工作方式：

class FirstNameOnly(Base):
    # ...

    first_name: Mapped[str]

    @hybrid_property
    def name(self) -> str:
        return self.first_name

    @name.inplace.setter
    def _name_setter(self, value: str) -> None:
        self.first_name = value

class FirstNameLastName(FirstNameOnly):
    # ...

    last_name: Mapped[str]

    # 'inplace' is not used here; calling getter creates a copy
    # of FirstNameOnly.name that is local to FirstNameLastName
    @FirstNameOnly.name.getter
    def name(self) -> str:
        return self.first_name + ' ' + self.last_name

    @name.inplace.setter
    def _name_setter(self, value: str) -> None:
        self.first_name, self.last_name = value.split(' ', 1)

上面，FirstNameLastName类引用了从FirstNameOnly.name到混合的混合，以重新用于子类的 getter 和 setter。

当仅覆盖hybrid_property.expression()和hybrid_property.comparator()作为对超类的第一引用时，这些名称与在类级别返回的类级别QueryableAttribute对象上的同名访问器发生冲突。要在直接引用父类描述符时覆盖这些方法，请添加特殊限定符hybrid_property.overrides，它将将被仪器化的属性反向引用回混合对象：

class FirstNameLastName(FirstNameOnly):
    # ...

    last_name: Mapped[str]

    @FirstNameOnly.name.overrides.expression
    @classmethod
    def name(cls):
        return func.concat(cls.first_name, ' ', cls.last_name)

混合值对象

请注意，在我们之前的示例中，如果我们将SearchWord实例的word_insensitive属性与普通的 Python 字符串进行比较，普通的 Python 字符串不会被强制转换为小写 - 我们构建的CaseInsensitiveComparator，由@word_insensitive.comparator返回，仅适用于 SQL 端。

自定义比较器的更全面形式是构建一个混合值对象。这种技术将目标值或表达式应用于一个值对象，然后该值对象在所有情况下由访问器返回。值对象允许控制对值的所有操作以及如何处理比较值，无论是在 SQL 表达式端还是 Python 值端。用新的CaseInsensitiveWord类替换之前的CaseInsensitiveComparator类：

class CaseInsensitiveWord(Comparator):
    "Hybrid value representing a lower case representation of a word."

    def __init__(self, word):
        if isinstance(word, basestring):
            self.word = word.lower()
        elif isinstance(word, CaseInsensitiveWord):
            self.word = word.word
        else:
            self.word = func.lower(word)

    def operate(self, op, other, **kwargs):
        if not isinstance(other, CaseInsensitiveWord):
            other = CaseInsensitiveWord(other)
        return op(self.word, other.word, **kwargs)

    def __clause_element__(self):
        return self.word

    def __str__(self):
        return self.word

    key = 'word'
    "Label to apply to Query tuple results"

在上文中，CaseInsensitiveWord 对象表示 self.word，它可能是一个 SQL 函数，也可能是一个 Python 本地函数。通过重写 operate() 和 __clause_element__() 方法，以 self.word 为基础进行操作，所有比较操作都将针对 word 的“转换”形式进行，无论是在 SQL 还是 Python 方面。我们的 SearchWord 类现在可以通过单一的混合调用无条件地提供 CaseInsensitiveWord 对象：

class SearchWord(Base):
    __tablename__ = 'searchword'
    id: Mapped[int] = mapped_column(primary_key=True)
    word: Mapped[str]

    @hybrid_property
    def word_insensitive(self) -> CaseInsensitiveWord:
        return CaseInsensitiveWord(self.word)

word_insensitive 属性现在具有普遍的不区分大小写的比较行为，包括 SQL 表达式与 Python 表达式（请注意此处的 Python 值在 Python 侧被转换为小写）：

>>> print(select(SearchWord).filter_by(word_insensitive="Trucks"))
SELECT  searchword.id  AS  searchword_id,  searchword.word  AS  searchword_word
FROM  searchword
WHERE  lower(searchword.word)  =  :lower_1

SQL 表达式与 SQL 表达式：

>>> from sqlalchemy.orm import aliased
>>> sw1 = aliased(SearchWord)
>>> sw2 = aliased(SearchWord)
>>> print(
...     select(sw1.word_insensitive, sw2.word_insensitive).filter(
...         sw1.word_insensitive > sw2.word_insensitive
...     )
... )
SELECT  lower(searchword_1.word)  AS  lower_1,
lower(searchword_2.word)  AS  lower_2
FROM  searchword  AS  searchword_1,  searchword  AS  searchword_2
WHERE  lower(searchword_1.word)  >  lower(searchword_2.word)

Python 只有表达式：

>>> ws1 = SearchWord(word="SomeWord")
>>> ws1.word_insensitive == "sOmEwOrD"
True
>>> ws1.word_insensitive == "XOmEwOrX"
False
>>> print(ws1.word_insensitive)
someword

混合值模式对于任何可能具有多个表示形式的值非常有用，例如时间戳、时间间隔、测量单位、货币和加密密码等。

另请参阅

混合类型和值无关类型 - 在 techspot.zzzeek.org 博客上

值无关类型，第二部分 - 在 techspot.zzzeek.org 博客上

API 参考

对象名称	描述
Comparator	一个辅助类，允许轻松构建用于混合类型的自定义 `PropComparator` 类。
hybrid_method	允许定义具有实例级和类级行为的 Python 对象方法的装饰器。
hybrid_property	允许定义具有实例级和类级行为的 Python 描述符的装饰器。
HybridExtensionType	枚举类型。

class sqlalchemy.ext.hybrid.hybrid_method

允许定义具有实例级和类级行为的 Python 对象方法的装饰器。

成员

init(), expression(), extension_type, inplace, is_attribute

类签名

class sqlalchemy.ext.hybrid.hybrid_method (sqlalchemy.orm.base.InspectionAttrInfo, typing.Generic)

method __init__(func: Callable[[Concatenate[Any, _P]], _R], expr: Callable[[Concatenate[Any, _P]], SQLCoreOperations[_R]] | None = None)

创建一个新的 hybrid_method。

通常使用装饰器：

from sqlalchemy.ext.hybrid import hybrid_method

class SomeClass:
    @hybrid_method
    def value(self, x, y):
        return self._value + x + y

    @value.expression
    @classmethod
    def value(cls, x, y):
        return func.some_function(cls._value, x, y)

method expression(expr: Callable[[Concatenate[Any, _P]], SQLCoreOperations[_R]]) → hybrid_method[_P, _R]

提供一个修改装饰器，定义一个生成 SQL 表达式的方法。

attribute extension_type: InspectionAttrExtensionType = 'HYBRID_METHOD'

扩展类型（如果有）。默认为 NotExtension.NOT_EXTENSION

另请参阅

HybridExtensionType

AssociationProxyExtensionType

attribute inplace

返回此 hybrid_method 的 inplace mutator。

当调用 hybrid_method.expression() 装饰器时，hybrid_method 类已经执行“in place”变异，因此此属性返回 Self。

2.0.4 版中的新功能。

另请参阅

使用 inplace 创建符合 pep-484 标准的混合属性（#hybrid-pep484-naming）

attribute is_attribute = True

如果此对象是 Python 描述符，则为 True。

这可能是许多类型之一。通常是一个 QueryableAttribute，它代表一个 MapperProperty 上的属性事件。但也可以是一个扩展类型，例如 AssociationProxy 或 hybrid_property。InspectionAttr.extension_type 将指示一个常量，用于标识特定的子类型。

另请参阅

Mapper.all_orm_descriptors

class sqlalchemy.ext.hybrid.hybrid_property

一个装饰器，允许定义既有实例级别又有类级别行为的 Python 描述符。

成员

init(), comparator(), deleter(), expression(), extension_type, getter(), inplace, is_attribute, overrides, setter(), update_expression()

类签名

类sqlalchemy.ext.hybrid.hybrid_property（sqlalchemy.orm.base.InspectionAttrInfo, sqlalchemy.orm.base.ORMDescriptor)

method __init__(fget: _HybridGetterType[_T], fset: _HybridSetterType[_T] | None = None, fdel: _HybridDeleterType[_T] | None = None, expr: _HybridExprCallableType[_T] | None = None, custom_comparator: Comparator[_T] | None = None, update_expr: _HybridUpdaterType[_T] | None = None)

创建一个新的hybrid_property。

通常通过装饰器来使用：

from sqlalchemy.ext.hybrid import hybrid_property

class SomeClass:
    @hybrid_property
    def value(self):
        return self._value

    @value.setter
    def value(self, value):
        self._value = value

method comparator(comparator: _HybridComparatorCallableType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个自定义比较器生成方法。

被装饰方法的返回值应该是Comparator的一个实例。

注意

hybrid_property.comparator()装饰器替换了hybrid_property.expression()装饰器的使用。它们不能同时使用。

当在类级别调用混合属性时，此处给出的Comparator对象被包装在一个专门的QueryableAttribute中，这是 ORM 用来表示其他映射属性的相同类型的对象。这样做的原因是为了在返回的结构中保留其他类级别属性，如文档字符串和对混合属性本身的引用，而不对传入的原始比较器对象进行任何修改。

注意

当从拥有类引用混合属性时（例如SomeClass.some_hybrid），返回一个QueryableAttribute的实例，表示表达式或比较器对象作为这个混合对象。然而，该对象本身有名为expression和comparator的访问器；因此，在子类中尝试覆盖这些装饰器时，可能需要首先使用hybrid_property.overrides修饰符进行限定。有关详细信息，请参阅该修饰符。

method deleter(fdel: _HybridDeleterType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个删除方法。

method expression(expr: _HybridExprCallableType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个生成 SQL 表达式的方法。

当在类级别调用混合时，此处给出的 SQL 表达式将包装在一个专门的 QueryableAttribute 中，该对象与 ORM 用于表示其他映射属性的对象相同。这样做的原因是为了在返回的结构中保持其他类级别属性（如文档字符串和对混合本身的引用），而不对传入的原始 SQL 表达式进行任何修改。

注意

当从拥有类引用混合属性时（例如 SomeClass.some_hybrid），会返回一个 QueryableAttribute 的实例，表示表达式或比较器对象以及此混合对象。然而，该对象本身有名为 expression 和 comparator 的访问器；因此，在尝试在子类上覆盖这些装饰器时，可能需要首先使用 hybrid_property.overrides 修饰符进行限定。详情请参阅该修饰符。

参见

定义与属性行为不同的表达式行为

attribute extension_type: InspectionAttrExtensionType = 'HYBRID_PROPERTY'

扩展类型，如果有的话。默认为 NotExtension.NOT_EXTENSION

参见

HybridExtensionType

AssociationProxyExtensionType

method getter(fget: _HybridGetterType[_T]) → hybrid_property[_T]

提供一个修改装饰器，定义一个 getter 方法。

自 1.2 版新功能。

attribute inplace

返回此 hybrid_property 的 inplace mutator。

这是为了允许对混合进行原地变异，从而允许重用某个特定名称的第一个混合方法以添加更多方法，而无需将这些方法命名为相同的名称，例如：

class Interval(Base):
    # ...

    @hybrid_property
    def radius(self) -> float:
        return abs(self.length) / 2

    @radius.inplace.setter
    def _radius_setter(self, value: float) -> None:
        self.length = value * 2

    @radius.inplace.expression
    def _radius_expression(cls) -> ColumnElement[float]:
        return type_coerce(func.abs(cls.length) / 2, Float)

自 2.0.4 版新功能。

参见

使用 inplace 创建符合 pep-484 的混合属性

attribute is_attribute = True

如果此对象是 Python 描述符，则为 True。

这可以指代许多类型之一。通常是一个 QueryableAttribute，它代表一个 MapperProperty 的属性事件。但也可以是一个扩展类型，如 AssociationProxy 或 hybrid_property。InspectionAttr.extension_type 将引用一个常量，用于标识特定的子类型。

另请参见

Mapper.all_orm_descriptors

attribute overrides

用于覆盖现有属性的方法的前缀。

hybrid_property.overrides 访问器只是返回这个混合对象，当在父类的类级别调用时，将取消引用通常在此级别返回的“instrumented attribute”，并允许修改装饰器，如 hybrid_property.expression() 和 hybrid_property.comparator() 被使用，而不会与通常存在于 QueryableAttribute 上的同名属性发生冲突：

class SuperClass:
    # ...

    @hybrid_property
    def foobar(self):
        return self._foobar

class SubClass(SuperClass):
    # ...

    @SuperClass.foobar.overrides.expression
    def foobar(cls):
        return func.subfoobar(self._foobar)

版本 1.2 中新增。

另请参见

在子类中重用混合属性

method setter(fset: _HybridSetterType[_T]) → hybrid_property[_T]

提供一个定义 setter 方法的修改装饰器。

method update_expression(meth: _HybridUpdaterType[_T]) → hybrid_property[_T]

提供一个定义 UPDATE 元组生成方法的修改装饰器。

该方法接受一个值，该值将被渲染到 UPDATE 语句的 SET 子句中。然后该方法应将此值处理为适合最终 SET 子句的单独列表达式，并将它们作为 2 元组序列返回。每个元组包含一个列表达式作为键和要渲染的值。

例如：

class Person(Base):
    # ...

    first_name = Column(String)
    last_name = Column(String)

    @hybrid_property
    def fullname(self):
        return first_name + " " + last_name

    @fullname.update_expression
    def fullname(cls, value):
        fname, lname = value.split(" ", 1)
        return [
            (cls.first_name, fname),
            (cls.last_name, lname)
        ]

版本 1.2 中新增。

class sqlalchemy.ext.hybrid.Comparator

一个辅助类，允许轻松构建用于混合使用的自定义 PropComparator 类。

类签名

class sqlalchemy.ext.hybrid.Comparator (sqlalchemy.orm.PropComparator)

class sqlalchemy.ext.hybrid.HybridExtensionType

一个枚举。

成员

HYBRID_METHOD, HYBRID_PROPERTY

类签名

类sqlalchemy.ext.hybrid.HybridExtensionType（sqlalchemy.orm.base.InspectionAttrExtensionType）

attribute HYBRID_METHOD = 'HYBRID_METHOD'

表示一个InspectionAttr的符号，类型为hybrid_method

赋予InspectionAttr.extension_type属性。

另请参阅

Mapper.all_orm_attributes

attribute HYBRID_PROPERTY = 'HYBRID_PROPERTY'

表示一个InspectionAttr的符号

类型为hybrid_method。

赋予InspectionAttr.extension_type属性。

另请参阅

Mapper.all_orm_attributes

Indexable

原文：docs.sqlalchemy.org/en/20/orm/extensions/indexable.html

在 ORM 映射类上定义具有“index”属性的列的Indexable类型。

“index”表示属性与具有预定义索引以访问它的Indexable列的元素相关联。Indexable类型包括ARRAY、JSON和HSTORE等类型。

indexable扩展为任何Indexable类型的列的元素提供了类似于Column的接口。在简单情况下，它可以被视为一个Column - 映射属性。

概要

假设Person是一个具有主键和 JSON 数据字段的模型。虽然该字段可以包含任意数量的元素，但我们希望单独引用名为name的元素作为行为类似独立列的专用属性：

from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    name = index_property('data', 'name')

上面，name属性现在的行为类似于映射列。我们可以组合一个新的Person并设置name的值：

>>> person = Person(name='Alchemist')

现在值是可访问的：

>>> person.name
'Alchemist'

在幕后，JSON 字段被初始化为一个新的空字典，并设置了字段：

>>> person.data
{"name": "Alchemist'}

该字段是可变的：

>>> person.name = 'Renamed'
>>> person.name
'Renamed'
>>> person.data
{'name': 'Renamed'}

当使用index_property时，我们对可索引结构所做的更改也会自动跟踪为历史记录；我们不再需要使用MutableDict来跟踪此更改以进行工作单元。

删除也可以正常工作：

>>> del person.name
>>> person.data
{}

上面，删除person.name会删除字典中的值，但不会删除字典本身。

缺少键将产生AttributeError：

>>> person = Person()
>>> person.name
...
AttributeError: 'name'

除非设置默认值：

>>> class Person(Base):
>>>     __tablename__ = 'person'
>>>
>>>     id = Column(Integer, primary_key=True)
>>>     data = Column(JSON)
>>>
>>>     name = index_property('data', 'name', default=None)  # See default

>>> person = Person()
>>> print(person.name)
None

这些属性也可以在类级别访问。下面，我们演示了Person.name用于生成带有索引的 SQL 条件：

>>> from sqlalchemy.orm import Session
>>> session = Session()
>>> query = session.query(Person).filter(Person.name == 'Alchemist')

上述查询等效于：

>>> query = session.query(Person).filter(Person.data['name'] == 'Alchemist')

可以链接多个index_property对象以生成多层索引：

from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    birthday = index_property('data', 'birthday')
    year = index_property('birthday', 'year')
    month = index_property('birthday', 'month')
    day = index_property('birthday', 'day')

上面，一个查询如下：

q = session.query(Person).filter(Person.year == '1980')

在 PostgreSQL 后端上，上述查询将呈现为：

SELECT person.id, person.data
FROM person
WHERE person.data -> %(data_1)s -> %(param_1)s = %(param_2)s

默认值

index_property在索引的数据结构不存在时包含特殊行为，并且调用了一个设置操作：

对于给定整数索引值的index_property，默认的数据结构将是一个 Python 列表，其中包含至少与索引值一样多的None值；然后将该值设置到列表中的相应位置。这意味着对于索引值为零的情况，在设置给定值之前，列表将初始化为[None]，对于索引值为五的情况，在设置第五个元素之前，列表将初始化为[None, None, None, None, None]。请注意，现有的列表不会直接扩展以接收一个值。
对于给定任何其他类型的索引值（例如通常是字符串）的index_property，将使用 Python 字典作为默认数据结构。
可以使用index_property.datatype参数将默认数据结构设置为任何 Python 可调用对象，覆盖以前的规则。

子类化

index_property可以被子类化，特别是针对常见的提供值或 SQL 表达式强制转换的用例。以下是在使用 PostgreSQL JSON 类型时的常见用法，其中我们还希望包括自动转换加astext()：

class pg_json_property(index_property):
    def __init__(self, attr_name, index, cast_type):
        super(pg_json_property, self).__init__(attr_name, index)
        self.cast_type = cast_type

    def expr(self, model):
        expr = super(pg_json_property, self).expr(model)
        return expr.astext.cast(self.cast_type)

上述子类可以与 PostgreSQL 特定版本的JSON一起使用：

from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import JSON

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    age = pg_json_property('data', 'age', Integer)

实例级别的age属性与以前的工作方式相同；但是在渲染 SQL 时，将使用 PostgreSQL 的->>运算符进行索引访问，而不是通常的索引运算符->：

>>> query = session.query(Person).filter(Person.age < 20)

上述查询将呈现为：

SELECT person.id, person.data
FROM person
WHERE CAST(person.data ->> %(data_1)s AS INTEGER) < %(param_1)s

API 参考

对象名称	描述
index_property	属性生成器。生成的属性描述了一个与`Indexable`列相对应的对象属性。

class sqlalchemy.ext.indexable.index_property

属性生成器。生成的属性描述了一个与Indexable列相对应的对象属性。

另请参阅

sqlalchemy.ext.indexable

成员

init()

类签名

类 sqlalchemy.ext.indexable.index_property（sqlalchemy.ext.hybrid.hybrid_property）

method __init__(attr_name, index, default=<object object>, datatype=None, mutable=True, onebased=True)

创建一个新的 index_property。

参数：

attr_name – Indexable 类型列的属性名，或者返回可索引结构的其他属性。
index – 用于获取和设置此值的索引。这应该是整数的 Python 端索引值。
default – 当给定索引处没有值时，将返回的值。
datatype – 当字段为空时使用的默认数据类型。默认情况下，这是从使用的索引类型派生的；对于整数索引，是 Python 列表，对于任何其他类型的索引，是 Python 字典。对于列表，列表将初始化为长度至少为 index 的 None 值列表。
mutable – 如果为 False，则不允许对属性进行写入和删除。
onebased – 假设此值的 SQL 表示是基于一的；也就是说，SQL 中的第一个索引是 1，而不是零。

概要

假设 Person 是一个带有主键和 JSON 数据字段的模型。虽然此字段可以包含任意数量的元素，但我们希望单独引用称为 name 的元素，作为一个独立的属性，其行为类似于独立的列：

from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    name = index_property('data', 'name')

上面，name 属性现在的行为类似于映射列。我们可以组合一个新的 Person 并设置 name 的值：

>>> person = Person(name='Alchemist')

现在可以访问该值了：

>>> person.name
'Alchemist'

在幕后，JSON 字段被初始化为一个新的空字典，并设置了字段：

>>> person.data
{"name": "Alchemist'}

该字段是可变的：

>>> person.name = 'Renamed'
>>> person.name
'Renamed'
>>> person.data
{'name': 'Renamed'}

当使用 index_property 时，对可索引结构所做的更改也会自动跟踪为历史记录；我们不再需要使用 MutableDict 来跟踪此更改的工作单元。

删除操作也正常工作：

>>> del person.name
>>> person.data
{}

上面，对 person.name 的删除会删除字典中的值，但不会删除字典本身。

缺少的键会产生 AttributeError：

>>> person = Person()
>>> person.name
...
AttributeError: 'name'

除非设置了默认值：

>>> class Person(Base):
>>>     __tablename__ = 'person'
>>>
>>>     id = Column(Integer, primary_key=True)
>>>     data = Column(JSON)
>>>
>>>     name = index_property('data', 'name', default=None)  # See default

>>> person = Person()
>>> print(person.name)
None

这些属性也可以在类级别访问。下面，我们说明了 Person.name 用于生成带索引的 SQL 条件：

>>> from sqlalchemy.orm import Session
>>> session = Session()
>>> query = session.query(Person).filter(Person.name == 'Alchemist')

上面的查询等效于：

>>> query = session.query(Person).filter(Person.data['name'] == 'Alchemist')

可以链式连接多个 index_property 对象以产生多级索引：

from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    birthday = index_property('data', 'birthday')
    year = index_property('birthday', 'year')
    month = index_property('birthday', 'month')
    day = index_property('birthday', 'day')

上面，诸如以下查询：

q = session.query(Person).filter(Person.year == '1980')

在 PostgreSQL 后端上，上述查询将呈现为：

SELECT person.id, person.data
FROM person
WHERE person.data -> %(data_1)s -> %(param_1)s = %(param_2)s

默认值

index_property 包含了当索引数据结构不存在时的特殊行为，以及调用了设置操作时：

对于给定整数索引值的 index_property，默认数据结构将是包含 None 值的 Python 列表，至少与索引值一样长；然后将值设置在列表中的位置。这意味着对于索引值为零的索引值，列表将在设置给定值之前初始化为 [None]，而对于索引值为五的索引值，列表将在将第五个元素设置为给定值之前初始化为 [None, None, None, None, None]。请注意，现有列表不会在原地扩展以接收值。
对于给定任何其他类型的索引值（例如通常是字符串）的 index_property，将使用 Python 字典作为默认数据结构。
可以使用 index_property.datatype 参数将默认数据结构设置为任何 Python 可调用对象，从而覆盖之前的规则。

子类化

index_property 可以进行子类化，特别是用于提供在访问时进行值或 SQL 表达式强制转换的常见用例。下面是一个常见的配方，用于与 PostgreSQL JSON 类型一起使用，其中我们还希望包括自动转换以及 astext()：

class pg_json_property(index_property):
    def __init__(self, attr_name, index, cast_type):
        super(pg_json_property, self).__init__(attr_name, index)
        self.cast_type = cast_type

    def expr(self, model):
        expr = super(pg_json_property, self).expr(model)
        return expr.astext.cast(self.cast_type)

上述子类可与 PostgreSQL 特定版本的 JSON 一起使用：

from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import JSON

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    data = Column(JSON)

    age = pg_json_property('data', 'age', Integer)

在实例级别的 age 属性仍然可以正常工作；但是，在渲染 SQL 时，将使用 PostgreSQL 的 ->> 运算符进行索引访问，而不是通常的索引运算符 ->：

>>> query = session.query(Person).filter(Person.age < 20)

上述查询将呈现为：

SELECT person.id, person.data
FROM person
WHERE CAST(person.data ->> %(data_1)s AS INTEGER) < %(param_1)s

API 参考

对象名称	描述
index_property	一个属性生成器。生成的属性描述了一个与 `Indexable` 列对应的对象属性。

class sqlalchemy.ext.indexable.index_property

一个属性生成器。生成的属性描述了一个与 Indexable 列对应的对象属性。

另请参阅

sqlalchemy.ext.indexable

成员

init()

类签名

类 sqlalchemy.ext.indexable.index_property（sqlalchemy.ext.hybrid.hybrid_property）

method __init__(attr_name, index, default=<object object>, datatype=None, mutable=True, onebased=True)

创建一个新的 index_property。

参数：

attr_name – 一个可索引类型列的属性名称，或者返回可索引结构的其他属性。
index – 用于获取和设置此值的索引。这应该是整数的 Python 端索引值。
default – 在给定索引处没有值时返回的值。
datatype – 当字段为空时使用的默认数据类型。默认情况下，这是从使用的索引类型派生的；对于整数索引，是一个 Python 列表，对于任何其他类型的索引，是一个 Python 字典。对于列表，列表将被初始化为至少index元素长的 None 值列表。
mutable – 如果为 False，则将禁止对属性的写入和删除。
onebased – 假设此值的 SQL 表示是基于一的；也就是说，在 SQL 中，第一个索引是 1，而不是零。

替代类仪器化

原文：docs.sqlalchemy.org/en/20/orm/extensions/instrumentation.html

可扩展的类仪器化。

sqlalchemy.ext.instrumentation包提供了 ORM 内的类仪器化的替代系统。类仪器化是指 ORM 如何将属性放在类上，以维护数据并跟踪对该数据的更改，以及安装在类上的事件钩子。

注意

该扩展包是为了与其他已经执行自己仪器化的对象管理包集成而提供的。它不适用于一般用途。

查看仪器扩展如何使用的示例，请参见示例属性仪器化。

API 参考

对象名称	描述
ExtendedInstrumentationRegistry	使用额外的簿记扩展`InstrumentationFactory`，以适应多种类型的类管理器。
instrumentation_finders	一个可扩展的返回仪器化实现的可调用序列
INSTRUMENTATION_MANAGER	属性，当存在于映射类上时，选择自定义仪器化。
InstrumentationFactory	用于新的 ClassManager 实例的工厂。
InstrumentationManager	用户定义的类仪器化扩展。

sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER = '__sa_instrumentation_manager__'

属性，当存在于映射类上时，选择自定义仪器化。

允许类指定稍微或完全不同的技术来跟踪对映射属性和集合所做的更改。

在给定对象继承层次结构中只允许一个仪器化实现。

此属性的值必须是可调用的，并将传递一个类对象。可调用对象必须返回以下之一：

一个InstrumentationManager的实例或子类

实现所有或部分 InstrumentationManager 的对象（待办事项）

实现上述所有或部分的可调用对象字典（待办事项）

一个ClassManager的实例或子类

一旦导入了sqlalchemy.ext.instrumentation模块，此属性将被 SQLAlchemy 仪器解析所使用。如果在全局仪器查找器列表中安装了自定义查找器，则它们可能会选择是否尊重此属性。

class sqlalchemy.orm.instrumentation.InstrumentationFactory

用于新的 ClassManager 实例的工厂。

类签名

类sqlalchemy.orm.instrumentation.InstrumentationFactory (sqlalchemy.event.registry.EventTarget)

class sqlalchemy.ext.instrumentation.InstrumentationManager

用户定义的类仪器扩展。

InstrumentationManager 可以被子类化以改变类仪器化的进行方式。此类存在的目的是与其他对象管理框架集成，这些框架希望完全修改 ORM 的仪器方法，并且不适用于常规使用。有关截取类仪器化事件，请参阅InstrumentationEvents。

成员

dict_getter(), get_instance_dict(), initialize_instance_dict(), install_descriptor(), install_member(), install_state(), instrument_attribute(), instrument_collection_class(), manage(), manager_getter(), post_configure_attribute(), remove_state(), state_getter(), uninstall_descriptor(), uninstall_member(), unregister()

此类的 API 应被视为半稳定的，并且可能会随着新版本略微更改。

method dict_getter(class_)

method get_instance_dict(class_, instance)

method initialize_instance_dict(class_, instance)

method install_descriptor(class_, key, inst)

method install_member(class_, key, implementation)

method install_state(class_, instance, state)

method instrument_attribute(class_, key, inst)

method instrument_collection_class(class_, key, collection_class)

method manage(class_, manager)

method manager_getter(class_)

method post_configure_attribute(class_, key, inst)

method remove_state(class_, instance)

method state_getter(class_)

method uninstall_descriptor(class_, key)

method uninstall_member(class_, key)

method unregister(class_, manager)

sqlalchemy.ext.instrumentation.instrumentation_finders = [<function find_native_user_instrumentation_hook>]

一个可扩展的返回仪器实现的可调用序列

当一个类被注册时，每个可调用对象都将传递一个类对象。如果返回 None，则会查阅序列中的下一个查找器。否则，返回值必须是一个遵循与 sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER 相同指南的仪器工厂。

默认情况下，唯一的查找器是 find_native_user_instrumentation_hook，它搜索 INSTRUMENTATION_MANAGER。如果所有查找器都返回 None，则使用标准的 ClassManager 仪器。

class sqlalchemy.ext.instrumentation.ExtendedInstrumentationRegistry

用额外的记录扩展了 InstrumentationFactory，以适应多种类型的类管理器。

类签名

类 sqlalchemy.ext.instrumentation.ExtendedInstrumentationRegistry (sqlalchemy.orm.instrumentation.InstrumentationFactory)

API 参考

对象名称	描述
扩展仪器注册表	用额外的记录扩展了 `InstrumentationFactory`，以适应多种类型的类管理器。
instrumentation_finders	一个可扩展的序列，其中包含返回仪器实现的可调用对象。
INSTRUMENTATION_MANAGER	属性，在映射类上出现时选择自定义仪器。
仪器工厂	用于创建新的 ClassManager 实例的工厂。
仪器管理器	用户定义的类仪器扩展。

sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER = '__sa_instrumentation_manager__'

属性，在映射类上出现时选择自定义仪器。

允许一个类指定一种稍微或完全不同的技术来跟踪对映射属性和集合所做的更改。

在给定对象继承层次结构中只允许有一个仪器实现。

此属性的值必须是一个可调用对象，并将传递一个类对象。可调用对象必须返回以下之一：

InstrumentationManager 或其子类的实例

实现了所有或部分 InstrumentationManager 的对象（待办）

一个可调用对象的字典，实现了上述所有或部分功能（待办）

ClassManager 或其子类的实例

一旦导入 sqlalchemy.ext.instrumentation 模块，此属性将由 SQLAlchemy 仪器化解析所使用。如果在全局 instrumentation_finders 列表中安装了自定义查找器，则它们可能会选择是否尊重此属性。

class sqlalchemy.orm.instrumentation.InstrumentationFactory

生成新的 ClassManager 实例的工厂。

类签名

类 sqlalchemy.orm.instrumentation.InstrumentationFactory (sqlalchemy.event.registry.EventTarget)

class sqlalchemy.ext.instrumentation.InstrumentationManager

用户自定义类仪器化扩展。

可以通过子类化 InstrumentationManager 来更改类仪器化的方式。此类存在的目的是为了与其他希望完全修改 ORM 的仪器化方法的对象管理框架集成，并不适用于常规使用。要拦截类仪器化事件，请参阅 InstrumentationEvents。

成员

该类的 API 应被视为半稳定，可能会在新版本中略微更改。

method dict_getter(class_)

method get_instance_dict(class_, instance)

method initialize_instance_dict(class_, instance)

method install_descriptor(class_, key, inst)

method install_member(class_, key, implementation)

method install_state(class_, instance, state)

method instrument_attribute(class_, key, inst)

method instrument_collection_class(class_, key, collection_class)

method manage(class_, manager)

method manager_getter(class_)

method post_configure_attribute(class_, key, inst)

method remove_state(class_, instance)

method state_getter(class_)

method uninstall_descriptor(class_, key)

method uninstall_member(class_, key)

method unregister(class_, manager)

sqlalchemy.ext.instrumentation.instrumentation_finders = [<function find_native_user_instrumentation_hook>]

一个可扩展的可调用序列，返回仪器化实现。

当一个类被注册时，每个可调用对象都将传递一个类对象。如果返回 None，则会查阅序列中的下一个查找器。否则，返回值必须是一个遵循与 sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER 相同指南的检测工厂。

默认情况下，唯一的查找器是 find_native_user_instrumentation_hook，它搜索 INSTRUMENTATION_MANAGER。如果所有查找器都返回 None，则使用标准的 ClassManager 仪器化。

class sqlalchemy.ext.instrumentation.ExtendedInstrumentationRegistry

通过额外的簿记扩展InstrumentationFactory，以适应多种类型的类管理器。

类签名

类sqlalchemy.ext.instrumentation.ExtendedInstrumentationRegistry（sqlalchemy.orm.instrumentation.InstrumentationFactory）

ORM 示例

原文：docs.sqlalchemy.org/en/20/orm/examples.html

SQLAlchemy 发行版包含各种代码示例，展示了一组选择的模式，有些是典型的，有些则不太典型。所有示例都可以运行，并且可以在发行版的/examples目录中找到。这里可以找到所有示例的描述和源代码。

其他 SQLAlchemy 示例，一些是用户贡献的，可以在www.sqlalchemy.org/trac/wiki/UsageRecipes的 wiki 上找到。

映射配方

邻接列表

一个使用邻接列表模型映射的字典-字典结构的示例。

例如：

node = TreeNode('rootnode')
node.append('node1')
node.append('node3')
session.add(node)
session.commit()

dump_tree(node)

文件列表：

adjacency_list.py ### 关联

展示了“关联对象”模式的使用示例，其中一个中间类在两个关联在多对多模式中的类之间进行关联。

文件列表：

proxied_association.py - 与 basic_association 相同的示例，增加了对sqlalchemy.ext.associationproxy的使用，以使对OrderItem的显式引用变为可选。
basic_association.py - 演示了“Order”和一组“Item”对象之间的多对多关系，通过名为“OrderItem”的关联对象将每个购买价格关联起来。
dict_of_sets_with_default.py - 一个高级的关联代理示例，演示了关联代理的嵌套，以生成多级 Python 集合，本例中是一个具有字符串键和整数集合值的字典，隐藏了底层映射类。 ### Asyncio 集成

展示 SQLAlchemy 的 asyncio 引擎特性的示例。

文件列表：

async_orm.py - 演示了sqlalchemy.ext.asyncio.AsyncSession对象用于异步 ORM 使用。
async_orm_writeonly.py - 演示了使用只写关系来更简单地处理异步 ORM 集合。
gather_orm_statements.py - 演示了如何使用asyncio.gather()在许多 asyncio 数据库连接上并发运行多个语句，将 ORM 结果合并到单个AsyncSession中。
basic.py - 演示了 asyncio 引擎/连接接口。
greenlet_orm.py - 演示了使用 sqlalchemy.ext.asyncio.AsyncSession 对象进行异步 ORM 使用，包括可选的 run_sync() 方法。 ### 有向图

有向图结构的持久性示例。图被存储为一组边，每条边都引用节点表中的“下级”和“上级”节点。演示了基本的持久性和查询下级和上级邻居的方法：

n2 = Node(2)
n5 = Node(5)
n2.add_neighbor(n5)
print(n2.higher_neighbors())

文件清单：

directed_graph.py ### 作为字典的动态关系

演示如何在“动态”关系之上放置类似字典的外观，以便字典操作（假设简单字符串键）可以在不一次加载完整集合的情况下操作大集合。

文件清单：

dynamic_dict.py ### 通用关联

演示了将多种类型的父对象与特定子对象关联的各种方法。

所有示例都使用声明性扩展和声明性混合类。每个示例最后都呈现相同的用例 - 两个类，Customer 和 Supplier，都是 HasAddresses 混合类的子类，该混合类确保父类提供一个包含 Address 对象的 addresses 集合。

discriminator_on_association.py 和 generic_fk.py 脚本是在 2007 年博客文章使用 SQLAlchemy 实现多态关联中提出的配方的现代化版本。

文件清单：

table_per_association.py - 演示了一个提供通用关联的混合类，通过为每个父类生成单独的关联表来实现。关联的对象本身存储在所有父类共享的单个表中。
table_per_related.py - 演示了一种通用关联，将关联对象持久化在各自的表中，每个表都生成用于代表特定父类持久化这些对象。
discriminator_on_association.py - 演示了一个提供通用关联的混合类，使用单个目标表和单个关联表，所有父表都引用该关联表。关联表包含一个“鉴别器”列，确定每个特定行与哪种类型的父对象关联。
generic_fk.py - 展示了所谓的“通用外键”，类似于流行框架（如 Django、ROR 等）的方式。这种方法绕过了标准的引用完整性实践，因为“外键”列实际上不被约束以引用任何特定的表；相反，在应用程序逻辑中使用逻辑来确定引用的是哪个表。### 物化路径

使用 SQLAlchemy ORM 展示了用于分层数据的“物化路径”模式。

文件列表：

materialized_paths.py - 说明了“物化路径”模式。### 嵌套集

使用 SQLAlchemy ORM 展示了一种实现用于分层数据的“嵌套集”模式的简单方法。

文件列表：

nested_sets.py - Celko 的“嵌套集”树结构。### 性能

一个用于各种 SQLAlchemy 使用情况的性能分析套件。

每个套件都专注于特定的用例，具有特定的性能配置文件和相关影响：

批量插入
单个插入，有或没有事务
获取大量行
运行大量的短查询

所有套件包括一系列使用模式，既展示了核心使用，也展示了 ORM 使用，并且通常按性能从最差到最好的顺序排序，根据 SQLAlchemy 提供的功能量的多少相反排序，从最多到最少（这两件事通常完美对应）。

一个命令行工具在包级别被呈现，允许单独的套件运行：

$ python -m examples.performance --help
usage: python -m examples.performance [-h] [--test TEST] [--dburl DBURL]
                                      [--num NUM] [--profile] [--dump]
                                      [--echo]

                                      {bulk_inserts,large_resultsets,single_inserts}

positional arguments:
  {bulk_inserts,large_resultsets,single_inserts}
                        suite to run

optional arguments:
  -h, --help            show this help message and exit
  --test TEST           run specific test name
  --dburl DBURL         database URL, default sqlite:///profile.db
  --num NUM             Number of iterations/items/etc for tests;
                        default is module-specific
  --profile             run profiling and dump call counts
  --dump                dump full call profile (implies --profile)
  --echo                Echo SQL output

一个示例运行如下：

$ python -m examples.performance bulk_inserts

或者使用选项：

$ python -m examples.performance bulk_inserts \
    --dburl mysql+mysqldb://scott:tiger@localhost/test \
    --profile --num 1000

另请参阅

如何对 SQLAlchemy 驱动的应用程序进行性能分析？

文件列表

文件列表：

bulk_updates.py - 这一系列的测试将说明不同的方法来批量更新大量行（正在建设中！目前只有一个测试）
large_resultsets.py - 在这一系列的测试中，我们关注的是加载大量非常小而简单的行的时间。
bulk_inserts.py - 这一系列的测试说明了不同的方法来批量插入大量行。
short_selects.py - 这一系列的测试说明了通过主键选择单个记录的不同方法
single_inserts.py - 在这一系列的测试中，我们关注的是一种在独立事务内插入一行数据的方法，然后返回到基本上“关闭”的状态。这类似于启动数据库连接的 API 调用，插入行，提交并关闭。
main.py - 允许将 examples/performance 包作为脚本运行。

使用时间运行所有测试

这是默认形式的运行：

$ python -m examples.performance single_inserts
Tests to run: test_orm_commit, test_bulk_save,
              test_bulk_insert_dictionaries, test_core,
              test_core_query_caching, test_dbapi_raw_w_connect,
              test_dbapi_raw_w_pool

test_orm_commit : Individual INSERT/COMMIT pairs via the
    ORM (10000 iterations); total time 13.690218 sec
test_bulk_save : Individual INSERT/COMMIT pairs using
    the "bulk" API  (10000 iterations); total time 11.290371 sec
test_bulk_insert_dictionaries : Individual INSERT/COMMIT pairs using
    the "bulk" API with dictionaries (10000 iterations);
    total time 10.814626 sec
test_core : Individual INSERT/COMMIT pairs using Core.
    (10000 iterations); total time 9.665620 sec
test_core_query_caching : Individual INSERT/COMMIT pairs using Core
    with query caching (10000 iterations); total time 9.209010 sec
test_dbapi_raw_w_connect : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection each time (10000 iterations); total time 9.551103 sec
test_dbapi_raw_w_pool : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection pool (10000 iterations); total time 8.001813 sec

为单个测试生成配置文件

可以为所有测试或更常见的是单个测试生成 Python 配置文件输出：

$ python -m examples.performance single_inserts --test test_core --num 1000 --dump
Tests to run: test_core
test_core : Individual INSERT/COMMIT pairs using Core. (1000 iterations); total fn calls 186109
         186109 function calls (186102 primitive calls) in 1.089 seconds

   Ordered by: internal time, call count

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    0.634    0.001    0.634    0.001 {method 'commit' of 'sqlite3.Connection' objects}
     1000    0.154    0.000    0.154    0.000 {method 'execute' of 'sqlite3.Cursor' objects}
     1000    0.021    0.000    0.074    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/sql/compiler.py:1950(_get_colparams)
     1000    0.015    0.000    0.034    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/engine/default.py:503(_init_compiled)
        1    0.012    0.012    1.091    1.091 examples/performance/single_inserts.py:79(test_core)

    ...

编写您自己的套件

分析套件系统是可扩展的，并且可以应用于您自己的一组测试。这是一个有价值的技术，可用于决定一些性能关键的程序集的正确方法。例如，如果我们想要分析几种加载之间的差异，我们可以创建一个名为test_loads.py的文件，内容如下：

from examples.performance import Profiler
from sqlalchemy import Integer, Column, create_engine, ForeignKey
from sqlalchemy.orm import relationship, joinedload, subqueryload, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = None
session = None

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))

# Init with name of file, default number of items
Profiler.init("test_loads", 1000)

@Profiler.setup_once
def setup_once(dburl, echo, num):
    "setup once.  create an engine, insert fixture data"
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    sess = Session(engine)
    sess.add_all([
        Parent(children=[Child() for j in range(100)])
        for i in range(num)
    ])
    sess.commit()

@Profiler.setup
def setup(dburl, echo, num):
    "setup per test.  create a new Session."
    global session
    session = Session(engine)
    # pre-connect so this part isn't profiled (if we choose)
    session.connection()

@Profiler.profile
def test_lazyload(n):
    "load everything, no eager loading."

    for parent in session.query(Parent):
        parent.children

@Profiler.profile
def test_joinedload(n):
    "load everything, joined eager loading."

    for parent in session.query(Parent).options(joinedload("children")):
        parent.children

@Profiler.profile
def test_subqueryload(n):
    "load everything, subquery eager loading."

    for parent in session.query(Parent).options(subqueryload("children")):
        parent.children

if __name__ == '__main__':
    Profiler.main()

我们可以直接运行我们的新脚本：

$ python test_loads.py  --dburl postgresql+psycopg2://scott:tiger@localhost/test
Running setup once...
Tests to run: test_lazyload, test_joinedload, test_subqueryload
test_lazyload : load everything, no eager loading. (1000 iterations); total time 11.971159 sec
test_joinedload : load everything, joined eager loading. (1000 iterations); total time 2.754592 sec
test_subqueryload : load everything, subquery eager loading. (1000 iterations); total time 2.977696 sec
```  ### 太空入侵者

使用 SQLite 作为状态机的太空入侵者游戏。

最初于 2012 年开发。适应在 Python 3 中工作。

在文本控制台中使用 ASCII 艺术运行。

![../_images/space_invaders.jpg](https://gitee.com/OpenDocCN/py-docs-zh/raw/master/docs/sqlalch_20/img/df5d8744e7ec946672bdde78d1db980c.png)

要运行：

```py
python -m examples.space_invaders.space_invaders

在运行时，观察日志中的 SQL 输出：

tail -f space_invaders.log

祝您愉快！

文件列表：

space_invaders.py ### 对象版本控制

使用历史表进行版本控制

说明了一个扩展，它为实体创建版本表并为每个更改存储记录。给定的扩展生成一个匿名的“history”类，表示目标对象的历史版本。

与使用时间行进行版本控制示例相比，该示例将更新写入为同一表中的新行，而不使用单独的历史表。

通过一个名为test_versioning.py的单元测试模块演示了用法，该模块可以像其他模块一样运行，内部使用unittest：

python -m examples.versioned_history.test_versioning

示例用法的片段，使用声明性：

from history_meta import Versioned, versioned_session

class Base(DeclarativeBase):
    pass

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

Session = sessionmaker(bind=engine)
versioned_session(Session)

sess = Session()
sc = SomeClass(name='sc1')
sess.add(sc)
sess.commit()

sc.name = 'sc1modified'
sess.commit()

assert sc.version == 2

SomeClassHistory = SomeClass.__history_mapper__.class_

assert sess.query(SomeClassHistory).\
            filter(SomeClassHistory.version == 1).\
            all() \
            == [SomeClassHistory(version=1, name='sc1')]

Versioned混合类设计用于与声明性结合使用。要将该扩展与经典映射器一起使用，可以应用_history_mapper函数：

from history_meta import _history_mapper

m = mapper(SomeClass, sometable)
_history_mapper(m)

SomeHistoryClass = SomeClass.__history_mapper__.class_

版本控制示例还与 ORM 乐观并发特性集成，文档化在配置版本计数器中。要启用此功能，请将标志Versioned.use_mapper_versioning设置为 True：

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    use_mapper_versioning = True

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

如果两个具有相同版本标识符的SomeClass实例被同时更新并发送到数据库进行并发 UPDATE，如果数据库隔离级别允许两个 UPDATE 语句继续，其中一个将失败，因为它不再针对最后已知的版本标识符。

文件列表：

test_versioning.py - 展示history_meta.py模块函数用法的单元测试。
history_meta.py - 版本控制混合类和其他实用程序。 #### 使用时间行进行版本控制

几个示例说明了拦截首先被解释为对行的 UPDATE 的更改的技术，并将其转换为对新行的 INSERT，将先前的行保留为历史版本。

与带有历史表的版本化示例进行比较，该示例将历史行写入单独的历史表。

文件列表：

versioned_rows.py - 展示了拦截对象更改的方法，将对单个行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，保持旧行不变。
versioned_rows_w_versionid.py - 展示了拦截对象更改的方法，将对单个行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，保持旧行不变。
versioned_map.py - 围绕“垂直表”结构的概念构建的 versioned_rows 示例的变体，类似于垂直属性映射示例中所示的那些。
versioned_update_old_row.py - 展示了versioned_rows.py中相同的 UPDATE 转换为 INSERT 技术，但还发出了一个 UPDATE 命令来影响旧行的时间戳更改。还包括一个SessionEvents.do_orm_execute()钩子，以限制查询仅针对最新版本。 ### 垂直属性映射

展示了“垂直表”映射。

“垂直表”是指将对象的各个属性存储为表中的不同行的技术。 “垂直表”技术用于持久化可以具有各种属性集的对象，但牺牲了简单的查询控制和简洁性。它通常在内容/文档管理系统中找到，以灵活地表示用户创建的结构。

给出了两种方法。在第二种方法中，每行引用一个“数据类型”，其中包含有关属性中存储的信息类型的信息，例如整数、字符串或日期。

例子：

shrew = Animal(u'shrew')
shrew[u'cuteness'] = 5
shrew[u'weasel-like'] = False
shrew[u'poisonous'] = True

session.add(shrew)
session.flush()

q = (session.query(Animal).
     filter(Animal.facts.any(
       and_(AnimalFact.key == u'weasel-like',
            AnimalFact.value == True))))
print('weasel-like animals', q.all())

文件列表：

dictlike-polymorphic.py - 将多态值的垂直表映射为字典。
dictlike.py - 将垂直表映射为字典。 ## 继承映射配方

基本继承映射

单表、联接表和具体表继承的工作示例，如映射类继承层次结构中所述。

文件列表：

joined.py - 联接表（每个子类一个表）继承示例。
concrete.py - 具体表（基于类的表）继承示例。
single.py - 单表（基于层次结构的表）继承示例。

特殊 API

属性仪器化

示例说明了对 SQLAlchemy 属性管理系统的修改。

文件列表：

listen_for_events.py - 展示了如何将事件附加到所有被检测的属性，并监听更改事件。
active_column_defaults.py - 说明了如何使用AttributeEvents.init_scalar()事件，配合核心列默认值，以提供 ORM 对象，当访问未设置的属性时自动产生默认值。
custom_management.py - 说明了如何使用sqlalchemy.ext.instrumentation扩展包自定义类仪器化。### 水平分片

SQLAlchemy 分片 API 的基本示例。分片是指在多个数据库之间水平扩展数据。

“分片”映射的基本组件包括：

多个Engine实例，每个都分配了一个“分片 id”。这些Engine实例可能引用不同的数据库，或者同一数据库中的不同模式/帐户，或者它们甚至可以仅通过会导致它们在使用时访问不同模式或表的选项进行区分。
一个函数，它可以根据要保存的实例返回单个分片 id；这称为“shard_chooser”。
一个可以返回适用于特定实例标识符的分片 id 列表的函数；这称为“id_chooser”。如果返回所有分片 id，则将搜索所有分片。
一个函数，它可以根据特定查询返回要尝试的分片 id 列表（“query_chooser”）。如果返回所有分片 id，则将查询所有分片并将结果连接在一起。

在这些示例中，使用不同类型的分片对相同的基本示例进行操作，该示例根据每个大陆的天气数据进行处理。我们提供了示例的 shard_chooser、id_chooser 和 query_chooser 函数。query_chooser 说明了检查 SQL 表达式元素以尝试确定所请求的单个分片。

创建通用分片例程是组织多个数据库实例的问题的一种雄心勃勃的方法。对于更简明的替代方案，“不同实体”方法是一种以明确方式将对象分配给不同表（以及可能的数据库节点）的简单方法 - 在EntityName的维基页面中有描述。

文件列表：

separate_databases.py - 演示了使用不同的 SQLite 数据库进行分片。
separate_tables.py - 演示了使用单个 SQLite 数据库进行分片，但是会使用命名约定来创建多个表。
separate_schema_translates.py - 演示了在使用具有多个模式的单个数据库进行分片时，可以为每个分片使用不同的“schema_translates_map”。
asyncio.py - 演示了与 asyncio 一起使用的分片 API。

扩展 ORM

ORM 查询事件

演示了如何使用 dogpile.cache 功能嵌入 ORM 查询，允许完全控制缓存以及从长期缓存中提取“延迟加载”的属性。

示例包括演示with_loader_criteria() 选项以及SessionEvents.do_orm_execute() 钩子的用法。

从 SQLAlchemy 1.4 开始，Query 构造与 Select 构造合并在一起，因此这两个对象基本上是相同的。

文件列表：

temporal_range.py - 演示了将应用于选定实体的自定义每个查询条件。
Dogpile 缓存

filter_public.py - 演示了应用于特定类型实体的全局条件。

在这个演示中，以下技术被说明：

使用 SessionEvents.do_orm_execute() 事件挂钩
绕过 Session.execute() 的基本技术，从自定义缓存源中获取数据，而不是从数据库中获取。
利用 dogpile.cache 进行基本缓存，使用“区域”允许对固定配置集合进行全局控制。
使用自定义的 UserDefinedOption 对象配置语句对象中的选项。

另请参阅

重新执行语句 - 包含此处提出的技术的一般示例。

例如：

# query for Person objects, specifying cache
stmt = select(Person).options(FromCache("default"))

# specify that each Person's "addresses" collection comes from
# cache too
stmt = stmt.options(RelationshipCache(Person.addresses, "default"))

# execute and results
result = session.execute(stmt)

print(result.scalars().all())

要运行，必须安装 SQLAlchemy 和 dogpile.cache，或者将它们安装到当前的 PYTHONPATH。演示将创建一个本地目录用于数据文件，插入初始数据，然后运行。第二次运行演示将利用已经存在的缓存文件，并且仅会发出一条 SQL 语句来查询两个表 - 但是显示的结果将利用数十个懒加载，所有懒加载都从缓存中获取。

演示脚本本身按复杂性顺序作为 Python 模块运行，以便相对导入起作用。

python -m examples.dogpile_caching.helloworld

python -m examples.dogpile_caching.relationship_caching

python -m examples.dogpile_caching.advanced

python -m examples.dogpile_caching.local_session_caching

文件列表：

environment.py - 确定数据/缓存文件路径和配置，如果需要，引导装置数据。
caching_query.py - 表示允许使用 SQLAlchemy 进行 Dogpile 缓存的函数和类。引入了一个称为 FromCache 的查询选项。
model.py - 数据模型，表示具有多个 Address 对象的 Person，每个对象都有 PostalCode、City、Country。
fixture_data.py - 安装一些示例数据。这里有一些美国/加拿大几个城市的少量邮政编码。然后，安装 100 个 Person 记录，每个记录都有一个随机选择的邮政编码。
helloworld.py - 演示如何加载一些数据，并缓存结果。
relationship_caching.py - 演示如何在关联端点上添加缓存选项，以便懒加载从缓存中加载。
advanced.py - 演示如何使用 Query 结合 FromCache 选项，包括前端加载、缓存失效和集合缓存。
local_session_caching.py - 此示例创建了一个新的 dogpile.cache 后端，将数据持久化在当前会话的字典中。移除会话后，缓存消失。

映射配方

邻接表

使用邻接表模型映射的字典-字典结构的示例。

例如：

node = TreeNode('rootnode')
node.append('node1')
node.append('node3')
session.add(node)
session.commit()

dump_tree(node)

文件列表：

adjacency_list.py ### 关联

描述了“关联对象”模式的使用示例，其中一个中介类在两个以多对多模式关联的类之间进行关系中介。

文件清单：

proxied_association.py - 与 basic_association 相同的示例，添加了对 sqlalchemy.ext.associationproxy 的使用，以使对 OrderItem 的显式引用是可选的。
basic_association.py - 演示了“订单”和一组“商品”对象之间的多对多关系，通过名为“OrderItem”的关联对象为每个关联购买价格。
dict_of_sets_with_default.py - 一个高级关联代理示例，演示了关联代理的嵌套以生成多级 Python 集合，本例中是一个具有字符串键和整数集合值的字典，它隐藏了底层映射类。 ### Asyncio 集成

描述了 SQLAlchemy 的 asyncio 引擎功能的示例。

文件清单：

async_orm.py - 演示了使用 sqlalchemy.ext.asyncio.AsyncSession 对象进行异步 ORM 使用。
async_orm_writeonly.py - 展示了使用只写关系来更简单地处理 asyncio 下的 ORM 集合。
gather_orm_statements.py - 演示了如何使用 asyncio.gather() 在许多 asyncio 数据库连接上同时运行许多语句，将 ORM 结果合并为单个 AsyncSession。
basic.py - 展示了 asyncio 引擎/连接接口。
greenlet_orm.py - 展示了使用 sqlalchemy.ext.asyncio.AsyncSession 对象进行异步 ORM 使用的示例，包括可选的 run_sync() 方法。 ### 有向图

有向图结构的持久性示例。图以一组边的形式存储，每个边都引用节点表中的“下限”和“上限”节点。演示了基本的持久性和查询“下限”和“上限”邻居的方法：

n2 = Node(2)
n5 = Node(5)
n2.add_neighbor(n5)
print(n2.higher_neighbors())

文件清单：

directed_graph.py ### 动态关系作为字典

展示了如何在“动态”关系之上放置类似字典的外观，以便字典操作（假设简单字符串键）可以在一次加载完整集合的情况下操作大型集合。

文件清单：

dynamic_dict.py ### 通用关联

展示了将多种类型的父对象与特定子对象关联的各种方法。

所有示例都使用了声明性扩展和声明性混合。每个示例最终呈现相同的用例 - 两个类，Customer和Supplier，都是HasAddresses混合类的子类，该混合类确保父类提供一个包含Address对象的addresses集合。

discriminator_on_association.py 和 generic_fk.py 脚本是 2007 年博客文章使用 SQLAlchemy 进行多态关联中提出的配方的现代化版本。

文件列表：

table_per_association.py - 展示了一个通过为每个父类单独生成关联表来提供通用关联的混合类。关联的对象本身存储在所有父类共享的单个表中。
table_per_related.py - 展示了一种通用关联，通过为每个父类生成单独的关联表来持久化关联对象，每个关联表都是为了代表特定父类而生成的。
discriminator_on_association.py - 展示了一个提供通用关联的混合类，使用单个目标表和单个关联表，所有父表都引用该关联表。关联表包含一个“鉴别器”列，用于确定每个关联表中的行与哪种类型的父对象相关联。
generic_fk.py - 展示了所谓的“通用外键”，类似于流行框架（如 Django、ROR 等）的做法。这种方法绕过了标准的参照完整性实践，因为“外键”列实际上并没有约束到任何特定的表；相反，应用程序逻辑用于确定引用的是哪个表。 ### 材料化路径

展示了使用 SQLAlchemy ORM 实现“材料化路径”模式的方法。

文件列表：

materialized_paths.py - 展示了“材料化路径”模式。 ### 嵌套集

展示了使用 SQLAlchemy ORM 实现“嵌套集”模式的基本方法。

文件列表：

nested_sets.py - Celko 的“嵌套集”树结构。 ### 性能

用于各种 SQLAlchemy 用例的性能分析套件。

每个套件专注于具有特定性能配置文件和相关影响的特定用例：

批量插入
单个插入，有或者没有事务
获取大量行
运行大量短查询

所有套件都包括一系列使用模式，说明了核心和 ORM 使用，并且通常按性能从最差到最佳的顺序排序，基于 SQLAlchemy 提供的功能数量，从最大到最小（这两个方面通常完美对应）。

一个命令行工具在包级别被呈现，它允许运行个别套件：

$ python -m examples.performance --help
usage: python -m examples.performance [-h] [--test TEST] [--dburl DBURL]
                                      [--num NUM] [--profile] [--dump]
                                      [--echo]

                                      {bulk_inserts,large_resultsets,single_inserts}

positional arguments:
  {bulk_inserts,large_resultsets,single_inserts}
                        suite to run

optional arguments:
  -h, --help            show this help message and exit
  --test TEST           run specific test name
  --dburl DBURL         database URL, default sqlite:///profile.db
  --num NUM             Number of iterations/items/etc for tests;
                        default is module-specific
  --profile             run profiling and dump call counts
  --dump                dump full call profile (implies --profile)
  --echo                Echo SQL output

示例运行如下：

$ python -m examples.performance bulk_inserts

或使用选项：

$ python -m examples.performance bulk_inserts \
    --dburl mysql+mysqldb://scott:tiger@localhost/test \
    --profile --num 1000

另请参阅

我如何分析使用 SQLAlchemy 的应用程序？

文件列表

文件列表：

bulk_updates.py - 这一系列的测试将演示不同的方式来批量更新大量行（正在建设中！目前只有一个测试）
large_resultsets.py - 在这一系列的测试中，我们正在研究加载大量非常小而简单行所需的时间。
bulk_inserts.py - 这一系列的测试演示了不同的方式来批量插入大量行。
short_selects.py - 这一系列的测试演示了不同的方式来通过主键选择单个记录
single_inserts.py - 在这一系列的测试中，我们正在研究一种在独立事务中插入一行数据的方法，然后返回到基本上是“关闭”的状态。这类似于一个启动数据库连接、插入行、提交并关闭的 API 调用。
main.py - 允许 examples/performance 包被当作脚本运行。

运行所有测试并计时

这是默认的运行形式：

$ python -m examples.performance single_inserts
Tests to run: test_orm_commit, test_bulk_save,
              test_bulk_insert_dictionaries, test_core,
              test_core_query_caching, test_dbapi_raw_w_connect,
              test_dbapi_raw_w_pool

test_orm_commit : Individual INSERT/COMMIT pairs via the
    ORM (10000 iterations); total time 13.690218 sec
test_bulk_save : Individual INSERT/COMMIT pairs using
    the "bulk" API  (10000 iterations); total time 11.290371 sec
test_bulk_insert_dictionaries : Individual INSERT/COMMIT pairs using
    the "bulk" API with dictionaries (10000 iterations);
    total time 10.814626 sec
test_core : Individual INSERT/COMMIT pairs using Core.
    (10000 iterations); total time 9.665620 sec
test_core_query_caching : Individual INSERT/COMMIT pairs using Core
    with query caching (10000 iterations); total time 9.209010 sec
test_dbapi_raw_w_connect : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection each time (10000 iterations); total time 9.551103 sec
test_dbapi_raw_w_pool : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection pool (10000 iterations); total time 8.001813 sec

为个别测试转储配置文件

Python 分析配置文件可以为所有测试或更常见的是为个别测试转储：

$ python -m examples.performance single_inserts --test test_core --num 1000 --dump
Tests to run: test_core
test_core : Individual INSERT/COMMIT pairs using Core. (1000 iterations); total fn calls 186109
         186109 function calls (186102 primitive calls) in 1.089 seconds

   Ordered by: internal time, call count

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    0.634    0.001    0.634    0.001 {method 'commit' of 'sqlite3.Connection' objects}
     1000    0.154    0.000    0.154    0.000 {method 'execute' of 'sqlite3.Cursor' objects}
     1000    0.021    0.000    0.074    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/sql/compiler.py:1950(_get_colparams)
     1000    0.015    0.000    0.034    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/engine/default.py:503(_init_compiled)
        1    0.012    0.012    1.091    1.091 examples/performance/single_inserts.py:79(test_core)

    ...

编写您自己的套件

分析套件系统是可扩展的，并且可以应用于您自己的一组测试。这是一个在决定某些性能关键程度的一组例程的正确方法时使用的宝贵技术。例如，如果我们想要分析几种加载之间的差异，我们可以创建一个名为test_loads.py的文件，内容如下：

from examples.performance import Profiler
from sqlalchemy import Integer, Column, create_engine, ForeignKey
from sqlalchemy.orm import relationship, joinedload, subqueryload, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = None
session = None

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))

# Init with name of file, default number of items
Profiler.init("test_loads", 1000)

@Profiler.setup_once
def setup_once(dburl, echo, num):
    "setup once.  create an engine, insert fixture data"
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    sess = Session(engine)
    sess.add_all([
        Parent(children=[Child() for j in range(100)])
        for i in range(num)
    ])
    sess.commit()

@Profiler.setup
def setup(dburl, echo, num):
    "setup per test.  create a new Session."
    global session
    session = Session(engine)
    # pre-connect so this part isn't profiled (if we choose)
    session.connection()

@Profiler.profile
def test_lazyload(n):
    "load everything, no eager loading."

    for parent in session.query(Parent):
        parent.children

@Profiler.profile
def test_joinedload(n):
    "load everything, joined eager loading."

    for parent in session.query(Parent).options(joinedload("children")):
        parent.children

@Profiler.profile
def test_subqueryload(n):
    "load everything, subquery eager loading."

    for parent in session.query(Parent).options(subqueryload("children")):
        parent.children

if __name__ == '__main__':
    Profiler.main()

我们可以直接运行我们的新脚本：

$ python test_loads.py  --dburl postgresql+psycopg2://scott:tiger@localhost/test
Running setup once...
Tests to run: test_lazyload, test_joinedload, test_subqueryload
test_lazyload : load everything, no eager loading. (1000 iterations); total time 11.971159 sec
test_joinedload : load everything, joined eager loading. (1000 iterations); total time 2.754592 sec
test_subqueryload : load everything, subquery eager loading. (1000 iterations); total time 2.977696 sec
```  ### 太空入侵

使用 SQLite 作为状态机的太空入侵游戏。

最初于 2012 年开发。适应 Python 3 中运行。

使用 ASCII 艺术在文本控制台中运行。

![../_images/space_invaders.jpg](https://gitee.com/OpenDocCN/py-docs-zh/raw/master/docs/sqlalch_20/img/df5d8744e7ec946672bdde78d1db980c.png)

要运行：

```py
python -m examples.space_invaders.space_invaders

在运行时，请观察日志中的 SQL 输出：

tail -f space_invaders.log

祝您愉快！

文件列表：

space_invaders.py ### 对象版本控制

使用历史表进行版本控制

演示了一个扩展，它为实体创建版本表并为每个更改存储记录。给定的扩展生成一个匿名的“历史”类，该类表示目标对象的历史版本。

与在相同表中将更新写为新行的使用时间行进行版本控制示例进行比较，而不使用单独的历史表。

通过一个单元测试模块 test_versioning.py 展示了用法，可以像运行任何其他模块一样运行，内部使用 unittest：

python -m examples.versioned_history.test_versioning

一个使用声明性的示例用法片段：

from history_meta import Versioned, versioned_session

class Base(DeclarativeBase):
    pass

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

Session = sessionmaker(bind=engine)
versioned_session(Session)

sess = Session()
sc = SomeClass(name='sc1')
sess.add(sc)
sess.commit()

sc.name = 'sc1modified'
sess.commit()

assert sc.version == 2

SomeClassHistory = SomeClass.__history_mapper__.class_

assert sess.query(SomeClassHistory).\
            filter(SomeClassHistory.version == 1).\
            all() \
            == [SomeClassHistory(version=1, name='sc1')]

Versioned 混合类设计用于与声明性一起使用。要将扩展与经典映射器一起使用，可以应用 _history_mapper 函数：

from history_meta import _history_mapper

m = mapper(SomeClass, sometable)
_history_mapper(m)

SomeHistoryClass = SomeClass.__history_mapper__.class_

版本控制示例还与文档中记录的 ORM 乐观并发特性集成在一起配置版本计数器。要启用此功能，请将标志 Versioned.use_mapper_versioning 设置为 True：

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    use_mapper_versioning = True

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

如果两个具有相同版本标识符的 SomeClass 实例被同时更新并发送到数据库以进行并发 UPDATE，如果数据库隔离级别允许两个 UPDATE 语句继续进行，则其中一个将失败，因为它不再针对最后已知的版本标识符。

文件列表：

test_versioning.py - 演示 history_meta.py 模块函数的用法的单元测试。
history_meta.py - 版本混合类和其他实用程序。 #### 使用时间行进行版本控制

有几个示例说明了拦截更改的技术，这些更改首先被解释为对行的 UPDATE，而实际上将其转换为对新行的 INSERT，使以前的行保持不变作为历史版本。

与将历史行写入单独的历史表的使用历史表进行版本控制示例进行比较。

文件列表：

versioned_rows.py - 演示拦截对象更改的方法，将单行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，保持旧行不变。
versioned_rows_w_versionid.py - 演示拦截对象更改的方法，将单行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，保持旧行不变。
versioned_map.py - 围绕“垂直表”结构的概念构建的 versioned_rows 示例的变体，类似于垂直属性映射示例中所示的那些。
versioned_update_old_row.py - 说明了versioned_rows.py中相同的 UPDATE 到 INSERT 技术，但也发出了对旧行的 UPDATE 以影响时间戳的更改。还包括一个SessionEvents.do_orm_execute()钩子，以限制查询仅限于最新版本。 ### 竖直属性映射

说明了“竖直表”映射。

“竖直表”是指一种技术，其中对象的各个属性被存储为表中的不同行。使用“竖直表”技术来持久化可以具有不同属性集的对象，但会牺牲简单的查询控制和简洁性。在内容/文档管理系统中通常可以灵活表示用户创建的结构。

给出了两种方法的变体。在第二种方法中，每行引用一个“数据类型”，其中包含关于属性中存储的信息类型的信息，例如整数、字符串或日期。

示例：

shrew = Animal(u'shrew')
shrew[u'cuteness'] = 5
shrew[u'weasel-like'] = False
shrew[u'poisonous'] = True

session.add(shrew)
session.flush()

q = (session.query(Animal).
     filter(Animal.facts.any(
       and_(AnimalFact.key == u'weasel-like',
            AnimalFact.value == True))))
print('weasel-like animals', q.all())

文件清单：

dictlike-polymorphic.py - 将具有多态值的竖直表映射为字典。
dictlike.py - 将竖直表映射为字典的示例。 ### 邻接表

以邻接表模型映射的字典嵌套结构的示例。

例如：

node = TreeNode('rootnode')
node.append('node1')
node.append('node3')
session.add(node)
session.commit()

dump_tree(node)

文件清单：

adjacency_list.py

关联

示例说明了“关联对象”模式的使用，其中一个中间类介于两个以多对多模式关联的类之间。

文件清单：

proxied_association.py - 与 basic_association 相同的示例，同时添加了对sqlalchemy.ext.associationproxy的使用，以使对OrderItem的显式引用成为可选。
basic_association.py - 说明了“订单”和“项目”对象集合之间的多对多关系，通过称为“OrderItem”的关联对象将每个订单价格关联起来。
dict_of_sets_with_default.py - 一个高级关联代理示例，说明了关联代理的嵌套以生成多级 Python 集合，本例中是一个具有字符串键和整数集合作为值的字典，该字典隐藏了底层的映射类。

Asyncio 集成

示例说明了 SQLAlchemy 的 asyncio 引擎功能。

文件清单：

async_orm.py - 演示了使用sqlalchemy.ext.asyncio.AsyncSession对象进行异步 ORM 使用。
async_orm_writeonly.py - 演示了在 asyncio 下使用只写关系来更简单地处理 ORM 集合。
gather_orm_statements.py - 演示了如何使用asyncio.gather()在许多 asyncio 数据库连接上并发运行多个语句，将 ORM 结果合并到单个AsyncSession中。
basic.py - 演示了 asyncio 引擎/连接接口。
greenlet_orm.py - 演示了使用 sqlalchemy.ext.asyncio.AsyncSession 对象进行异步 ORM 使用，包括可选的 run_sync()方法。

有向图

一个有向图结构的持久性示例。图被存储为一组边，每条边都引用节点表中的“较低”和“较高”节点。演示了基本的持久性和查询较低和较高邻居的方法：

n2 = Node(2)
n5 = Node(5)
n2.add_neighbor(n5)
print(n2.higher_neighbors())

文件清单：

directed_graph.py

动态关系作为字典

演示了如何在“动态”关系之上放置类似于字典的外观，以便字典操作（假设简单的字符串键）可以在大型集合上进行操作，而无需一次加载整个集合。

文件清单：

dynamic_dict.py

通用关联

演示了将多种类型的父类与特定子对象关联的各种方法。

所有示例都使用声明性扩展以及声明性 mixin。每个示例最后都呈现相同的用例 - 两个类，Customer和Supplier，都是HasAddresses mixin 的子类，该 mixin 确保父类提供了一个包含Address对象的addresses集合。

discriminator_on_association.py 和 generic_fk.py 脚本是 2007 年博客文章使用 SQLAlchemy 进行多态关联中提出的配方的现代化版本。

文件清单：

table_per_association.py - 通过为每个父类单独生成关联表格来提供通用关联的 mixin 示例。关联对象本身存储在所有父类之间共享的单个表中。
table_per_related.py - 演示了一种通用关联，它在各个表中持久化关联对象，每个表都生成来代表特定父类持久化这些对象。
discriminator_on_association.py - 演示了一种提供通用关联的 mixin，该关联使用单个目标表和单个关联表，所有父表都引用它。关联表包含一个“区分符”列，用于确定哪种类型的父对象与关联表中的每个特定行关联。
generic_fk.py - 演示了所谓的“通用外键”，类似于流行框架（如 Django，ROR 等）的方式。这种方法绕过了标准的参照完整性实践，因为“外键”列实际上并不限制引用任何特定表；相反，应用程序逻辑用于确定引用的是哪个表。

材料化路径

演示了使用 SQLAlchemy ORM 实现“材料化路径”模式的方法。

文件清单：

materialized_paths.py - 演示了“材料化路径”模式。

嵌套集

演示了使用 SQLAlchemy ORM 实现“嵌套集”模式的一种基本方法。

文件清单：

nested_sets.py - Celko 的“嵌套集”树结构。

性能

用于各种 SQLAlchemy 用例的性能分析套件。

每个套件都专注于特定用例，具有特定的性能概况和相关含义：

批量插入
单独插入，有或没有事务
获取大量行
运行大量短查询

所有套件都包括各种使用模式，说明了 Core 和 ORM 的使用，并且通常按性能从最差到最佳的顺序排序，根据 SQLAlchemy 提供的功能量的多少，从最大到最小排列（这两件事通常完美对应）。

一个命令行工具在包级别呈现，允许运行各个套件：

$ python -m examples.performance --help
usage: python -m examples.performance [-h] [--test TEST] [--dburl DBURL]
                                      [--num NUM] [--profile] [--dump]
                                      [--echo]

                                      {bulk_inserts,large_resultsets,single_inserts}

positional arguments:
  {bulk_inserts,large_resultsets,single_inserts}
                        suite to run

optional arguments:
  -h, --help            show this help message and exit
  --test TEST           run specific test name
  --dburl DBURL         database URL, default sqlite:///profile.db
  --num NUM             Number of iterations/items/etc for tests;
                        default is module-specific
  --profile             run profiling and dump call counts
  --dump                dump full call profile (implies --profile)
  --echo                Echo SQL output

一个示例运行如下：

$ python -m examples.performance bulk_inserts

或者带有选项：

$ python -m examples.performance bulk_inserts \
    --dburl mysql+mysqldb://scott:tiger@localhost/test \
    --profile --num 1000

另请参阅

我如何分析一个由 SQLAlchemy 驱动的应用程序？

文件清单

文件清单：

bulk_updates.py - 这一系列测试将说明不同的方式来批量更新大量行（正在施工！目前只有一个测试）
large_resultsets.py - 在这系列测试中，我们研究加载大量非常小而简单的行所需的时间。
bulk_inserts.py - 这一系列的测试展示了批量插入大量行的不同方法。
short_selects.py - 这一系列的测试展示了通过主键选择单个记录的不同方法。
single_inserts.py - 在这一系列的测试中，我们看到了一种在不同事务中插入行并且之后返回到实质上“关闭”状态的方法。这类似于启动数据库连接的 API 调用，插入行，提交并关闭的过程。
main.py - 允许 examples/performance 包作为脚本运行。

使用时间运行所有测试

这是运行的默认形式：

$ python -m examples.performance single_inserts
Tests to run: test_orm_commit, test_bulk_save,
              test_bulk_insert_dictionaries, test_core,
              test_core_query_caching, test_dbapi_raw_w_connect,
              test_dbapi_raw_w_pool

test_orm_commit : Individual INSERT/COMMIT pairs via the
    ORM (10000 iterations); total time 13.690218 sec
test_bulk_save : Individual INSERT/COMMIT pairs using
    the "bulk" API  (10000 iterations); total time 11.290371 sec
test_bulk_insert_dictionaries : Individual INSERT/COMMIT pairs using
    the "bulk" API with dictionaries (10000 iterations);
    total time 10.814626 sec
test_core : Individual INSERT/COMMIT pairs using Core.
    (10000 iterations); total time 9.665620 sec
test_core_query_caching : Individual INSERT/COMMIT pairs using Core
    with query caching (10000 iterations); total time 9.209010 sec
test_dbapi_raw_w_connect : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection each time (10000 iterations); total time 9.551103 sec
test_dbapi_raw_w_pool : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection pool (10000 iterations); total time 8.001813 sec

为个别测试输出简要概要

可以为所有测试或更常见的是个别测试转储 Python 分析输出：

$ python -m examples.performance single_inserts --test test_core --num 1000 --dump
Tests to run: test_core
test_core : Individual INSERT/COMMIT pairs using Core. (1000 iterations); total fn calls 186109
         186109 function calls (186102 primitive calls) in 1.089 seconds

   Ordered by: internal time, call count

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    0.634    0.001    0.634    0.001 {method 'commit' of 'sqlite3.Connection' objects}
     1000    0.154    0.000    0.154    0.000 {method 'execute' of 'sqlite3.Cursor' objects}
     1000    0.021    0.000    0.074    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/sql/compiler.py:1950(_get_colparams)
     1000    0.015    0.000    0.034    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/engine/default.py:503(_init_compiled)
        1    0.012    0.012    1.091    1.091 examples/performance/single_inserts.py:79(test_core)

    ...

编写你自己的测试套件

分析套件系统是可扩展的，并且可以应用于你自己的一组测试。这是一种在决定某些性能关键例程的正确方法时使用的有价值技术。例如，如果我们想要分析几种加载方式之间的差异，我们可以创建一个名为 test_loads.py 的文件，其内容如下：

from examples.performance import Profiler
from sqlalchemy import Integer, Column, create_engine, ForeignKey
from sqlalchemy.orm import relationship, joinedload, subqueryload, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = None
session = None

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))

# Init with name of file, default number of items
Profiler.init("test_loads", 1000)

@Profiler.setup_once
def setup_once(dburl, echo, num):
    "setup once.  create an engine, insert fixture data"
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    sess = Session(engine)
    sess.add_all([
        Parent(children=[Child() for j in range(100)])
        for i in range(num)
    ])
    sess.commit()

@Profiler.setup
def setup(dburl, echo, num):
    "setup per test.  create a new Session."
    global session
    session = Session(engine)
    # pre-connect so this part isn't profiled (if we choose)
    session.connection()

@Profiler.profile
def test_lazyload(n):
    "load everything, no eager loading."

    for parent in session.query(Parent):
        parent.children

@Profiler.profile
def test_joinedload(n):
    "load everything, joined eager loading."

    for parent in session.query(Parent).options(joinedload("children")):
        parent.children

@Profiler.profile
def test_subqueryload(n):
    "load everything, subquery eager loading."

    for parent in session.query(Parent).options(subqueryload("children")):
        parent.children

if __name__ == '__main__':
    Profiler.main()

我们可以直接运行我们的新脚本：

$ python test_loads.py  --dburl postgresql+psycopg2://scott:tiger@localhost/test
Running setup once...
Tests to run: test_lazyload, test_joinedload, test_subqueryload
test_lazyload : load everything, no eager loading. (1000 iterations); total time 11.971159 sec
test_joinedload : load everything, joined eager loading. (1000 iterations); total time 2.754592 sec
test_subqueryload : load everything, subquery eager loading. (1000 iterations); total time 2.977696 sec

文件列表

文件列表：

bulk_updates.py - 这一系列的测试将展示不同方法批量更新大量行（正在构建！目前只有一个测试）
large_resultsets.py - 在这一系列的测试中，我们关注的是加载大量非常小而简单的行的时间。
bulk_inserts.py - 这一系列的测试展示了批量插入大量行的不同方法。
short_selects.py - 这一系列的测试展示了通过主键选择单个记录的不同方法。
single_inserts.py - 在这一系列的测试中，我们看到了一种在不同事务中插入行并且之后返回到实质上“关闭”状态的方法。这类似于启动数据库连接的 API 调用，插入行，提交并关闭的过程。
main.py - 允许 examples/performance 包作为脚本运行。

使用时间运行所有测试

这是运行的默认形式：

$ python -m examples.performance single_inserts
Tests to run: test_orm_commit, test_bulk_save,
              test_bulk_insert_dictionaries, test_core,
              test_core_query_caching, test_dbapi_raw_w_connect,
              test_dbapi_raw_w_pool

test_orm_commit : Individual INSERT/COMMIT pairs via the
    ORM (10000 iterations); total time 13.690218 sec
test_bulk_save : Individual INSERT/COMMIT pairs using
    the "bulk" API  (10000 iterations); total time 11.290371 sec
test_bulk_insert_dictionaries : Individual INSERT/COMMIT pairs using
    the "bulk" API with dictionaries (10000 iterations);
    total time 10.814626 sec
test_core : Individual INSERT/COMMIT pairs using Core.
    (10000 iterations); total time 9.665620 sec
test_core_query_caching : Individual INSERT/COMMIT pairs using Core
    with query caching (10000 iterations); total time 9.209010 sec
test_dbapi_raw_w_connect : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection each time (10000 iterations); total time 9.551103 sec
test_dbapi_raw_w_pool : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection pool (10000 iterations); total time 8.001813 sec

为个别测试输出简要概要

可以为所有测试或更常见的是个别测试转储 Python 分析输出：

$ python -m examples.performance single_inserts --test test_core --num 1000 --dump
Tests to run: test_core
test_core : Individual INSERT/COMMIT pairs using Core. (1000 iterations); total fn calls 186109
         186109 function calls (186102 primitive calls) in 1.089 seconds

   Ordered by: internal time, call count

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    0.634    0.001    0.634    0.001 {method 'commit' of 'sqlite3.Connection' objects}
     1000    0.154    0.000    0.154    0.000 {method 'execute' of 'sqlite3.Cursor' objects}
     1000    0.021    0.000    0.074    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/sql/compiler.py:1950(_get_colparams)
     1000    0.015    0.000    0.034    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/engine/default.py:503(_init_compiled)
        1    0.012    0.012    1.091    1.091 examples/performance/single_inserts.py:79(test_core)

    ...

编写你自己的测试套件

性能分析套件系统是可扩展的，并且可以应用于您自己的一组测试。这是在决定某些性能关键例程的正确方法时使用的宝贵技术。例如，如果我们想要分析几种加载之间的差异，我们可以创建一个名为test_loads.py的文件，其中包含以下内容：

from examples.performance import Profiler
from sqlalchemy import Integer, Column, create_engine, ForeignKey
from sqlalchemy.orm import relationship, joinedload, subqueryload, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = None
session = None

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))

# Init with name of file, default number of items
Profiler.init("test_loads", 1000)

@Profiler.setup_once
def setup_once(dburl, echo, num):
    "setup once.  create an engine, insert fixture data"
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    sess = Session(engine)
    sess.add_all([
        Parent(children=[Child() for j in range(100)])
        for i in range(num)
    ])
    sess.commit()

@Profiler.setup
def setup(dburl, echo, num):
    "setup per test.  create a new Session."
    global session
    session = Session(engine)
    # pre-connect so this part isn't profiled (if we choose)
    session.connection()

@Profiler.profile
def test_lazyload(n):
    "load everything, no eager loading."

    for parent in session.query(Parent):
        parent.children

@Profiler.profile
def test_joinedload(n):
    "load everything, joined eager loading."

    for parent in session.query(Parent).options(joinedload("children")):
        parent.children

@Profiler.profile
def test_subqueryload(n):
    "load everything, subquery eager loading."

    for parent in session.query(Parent).options(subqueryload("children")):
        parent.children

if __name__ == '__main__':
    Profiler.main()

我们可以直接运行我们的新脚本：

$ python test_loads.py  --dburl postgresql+psycopg2://scott:tiger@localhost/test
Running setup once...
Tests to run: test_lazyload, test_joinedload, test_subqueryload
test_lazyload : load everything, no eager loading. (1000 iterations); total time 11.971159 sec
test_joinedload : load everything, joined eager loading. (1000 iterations); total time 2.754592 sec
test_subqueryload : load everything, subquery eager loading. (1000 iterations); total time 2.977696 sec

太空侵略者

使用 SQLite 作为状态机的太空侵略者游戏。

最初开发于 2012 年。已适配为在 Python 3 中运行。

在文本控制台中使用 ASCII 艺术运行。

../_images/space_invaders.jpg

运行：

python -m examples.space_invaders.space_invaders

在运行时，观察日志中的 SQL 输出：

tail -f space_invaders.log

尽情享受！

文件清单：

space_invaders.py

版本化对象

带有历史表的版本控制

演示了一个扩展，为实体创建版本表并存储每次更改的记录。给定的扩展生成一个匿名的“历史”类，表示目标对象的历史版本。

与使用时间行进行版本控制示例进行比较，该示例将更新写入为同一表中的新行，而不使用单独的历史表。

使用通过一个单元测试模块test_versioning.py进行演示，可以像其他模块一样运行，内部使用unittest：

python -m examples.versioned_history.test_versioning

一个示例用法片段，使用声明性：

from history_meta import Versioned, versioned_session

class Base(DeclarativeBase):
    pass

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

Session = sessionmaker(bind=engine)
versioned_session(Session)

sess = Session()
sc = SomeClass(name='sc1')
sess.add(sc)
sess.commit()

sc.name = 'sc1modified'
sess.commit()

assert sc.version == 2

SomeClassHistory = SomeClass.__history_mapper__.class_

assert sess.query(SomeClassHistory).\
            filter(SomeClassHistory.version == 1).\
            all() \
            == [SomeClassHistory(version=1, name='sc1')]

Versioned混合类设计用于与声明性一起使用。要将扩展与经典映射器一起使用，可以应用_history_mapper函数：

from history_meta import _history_mapper

m = mapper(SomeClass, sometable)
_history_mapper(m)

SomeHistoryClass = SomeClass.__history_mapper__.class_

版本示例还与 ORM 乐观并发功能集成，文档化在配置版本计数器。要启用此功能，请将标志Versioned.use_mapper_versioning设置为 True：

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    use_mapper_versioning = True

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

如果两个具有相同版本标识符的SomeClass实例被同时更新并发送到数据库进行并发更新，如果数据库隔离级别允许两个 UPDATE 语句继续进行，其中一个将失败，因为它不再针对最后已知的版本标识符。

文件清单：

test_versioning.py - 展示了history_meta.py模块函数的使用的单元测试。
history_meta.py - 带有版本控制的混合类和其他实用工具。#### 使用时间行进行版本控制

几个示例说明拦截更改的技术，这些更改首先被解释为对行的更新，而实际上将其转换为对新行的插入，保留先前的行作为历史版本。

与带有历史表的版本控制示例进行比较，该示例将历史行写入单独的历史表。

文件清单：

versioned_rows.py - 说明了拦截对象更改的方法，将对单个行的 UPDATE 语句转换为 INSERT 语句，以便使用新数据插入新行，保留旧行不变。
versioned_rows_w_versionid.py - 说明了拦截对象更改的方法，将对单个行的 UPDATE 语句转换为 INSERT 语句，以便使用新数据插入新行，保留旧行不变。
versioned_map.py - 围绕“垂直表”结构的概念构建的versioned_rows示例的变体，类似于垂直属性映射示例中所示的内容。
versioned_update_old_row.py - 说明了versioned_rows.py的相同 UPDATE 到 INSERT 技术，但还会对旧行进行 UPDATE 以影响时间戳的更改。还包括一个SessionEvents.do_orm_execute()挂钩来将查询限制为只有最新版本。 #### 使用历史表进行版本控制

展示了一个创建实体的版本表并为每个更改存储记录的扩展。给定的扩展生成一个匿名的“history”类，表示目标对象的历史版本。

与使用时间行版本化的例子相比，这些例子将更新写入相同表中的新行中，而不使用单独的历史表。

通过一个单元测试模块test_versioning.py来说明使用方法，可以像其他模块一样运行，内部使用unittest：

python -m examples.versioned_history.test_versioning

一个示例用法片段，使用声明式：

from history_meta import Versioned, versioned_session

class Base(DeclarativeBase):
    pass

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

Session = sessionmaker(bind=engine)
versioned_session(Session)

sess = Session()
sc = SomeClass(name='sc1')
sess.add(sc)
sess.commit()

sc.name = 'sc1modified'
sess.commit()

assert sc.version == 2

SomeClassHistory = SomeClass.__history_mapper__.class_

assert sess.query(SomeClassHistory).\
            filter(SomeClassHistory.version == 1).\
            all() \
            == [SomeClassHistory(version=1, name='sc1')]

Versioned mixin 设计为与声明性一起使用。要将扩展与经典映射器一起使用，可以应用_history_mapper函数：

from history_meta import _history_mapper

m = mapper(SomeClass, sometable)
_history_mapper(m)

SomeHistoryClass = SomeClass.__history_mapper__.class_

版本控制示例还与在配置版本计数器中记录的 ORM 乐观并发功能集成。要启用此功能，请将标志Versioned.use_mapper_versioning设置为 True：

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    use_mapper_versioning = True

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

上面，如果两个具有相同版本标识符的SomeClass实例被同时更新并发送到数据库进行并发更新，如果数据库隔离级别允许这两个 UPDATE 语句继续进行，其中一个将失败，因为它不再是针对最后已知版本标识符的。

文件列表：

test_versioning.py - 展示了history_meta.py模块函数的用法的单元测试。
history_meta.py - 版本混合类和其他实用工具。

使用时间行版本化

几个示例说明了拦截更改的技术，这些更改首先被解释为对行的 UPDATE，而实际上将其转换为对新行的 INSERT，将先前的行保留为历史版本。

与带有历史表的版本控制示例进行比较，该示例将历史行写入单独的历史表中。

文件列表：

versioned_rows.py - 展示了拦截对象更改的方法，将对单行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，同时保留旧行不变。
versioned_rows_w_versionid.py - 展示了拦截对象更改的方法，将对单行的 UPDATE 语句转换为 INSERT 语句，以便插入具有新数据的新行，同时保留旧行不变。
versioned_map.py - 围绕“垂直表”结构概念构建的 versioned_rows 示例的变体，类似于垂直属性映射示例中所示的那些。
versioned_update_old_row.py - 展示了与versioned_rows.py相同的 UPDATE 转换为 INSERT 技术，但还发出了对旧行的 UPDATE 以影响时间戳的更改。还包括一个SessionEvents.do_orm_execute()钩子，以限制查询仅针对最新版本。

垂直属性映射

展示了“垂直表”映射。

给出了两种方法的变体。在第二种方法中，每行引用一个包含有关存储在属性中的信息类型的“数据类型”，例如整数、字符串或日期。

示例：

shrew = Animal(u'shrew')
shrew[u'cuteness'] = 5
shrew[u'weasel-like'] = False
shrew[u'poisonous'] = True

session.add(shrew)
session.flush()

q = (session.query(Animal).
     filter(Animal.facts.any(
       and_(AnimalFact.key == u'weasel-like',
            AnimalFact.value == True))))
print('weasel-like animals', q.all())

文件列表：

dictlike-polymorphic.py - 将多态值的垂直表映射为字典。
dictlike.py - 将垂直表映射为字典。

继承映射配方

基本继承映射

单表、联表和具体表继承的工作示例，如映射类继承层次结构中所述。

文件列表：

joined.py - 联接表（每个子类一个表）继承示例。
concrete.py - 具体表（每类一表）继承示例。
single.py - 单表（按层次结构划分）继承示例。 ### 基本继承映射

作为 Mapping Class Inheritance Hierarchies 中描述的单表、连接表和具体表继承的工作示例。

文件清单：

joined.py - 连接表（每子类一表）继承示例。
concrete.py - 具体表（每类一表）继承示例。
single.py - 单表（按层次结构划分）继承示例。

特殊 API

属性仪器化

示例说明了对 SQLAlchemy 属性管理系统的修改。

文件清单：

listen_for_events.py - 演示了如何将事件附加到所有仪器化属性并监听更改事件。
active_column_defaults.py - 演示了如何使用AttributeEvents.init_scalar()事件，结合核心列默认值来提供 ORM 对象，当访问未设置的属性时自动产生默认值。
custom_management.py - 演示了如何使用sqlalchemy.ext.instrumentation扩展包进行自定义类仪器化。 ### 水平分片

使用 SQLAlchemy 分片 API 的基本示例。分片是指将数据横向扩展到多个数据库。

“分片”映射的基本组件包括：

多个Engine实例，每个实例分配一个“分片 ID”。这些Engine实例可以引用不同的数据库，或者同一数据库中的不同模式/帐户，或者甚至可以仅通过选项来区分，当使用时会使它们访问不同的模式或表。
给定要保存的实例，可以返回单个分片 ID 的函数；这称为“shard_chooser”。
可以返回适用于特定实例标识符的分片 ID 列表的函数；这称为“id_chooser”。如果返回所有分片 ID，则会搜索所有分片。
给定特定查询（“query_chooser”），可以返回要尝试的分片 ID 列表的函数。如果返回所有分片 ID，则会查询所有分片并将结果合并在一起。

在这些示例中，针对相同的基本示例使用不同类型的分片，适应以每个大陆为基础的天气数据。我们提供了示例的 shard_chooser、id_chooser 和 query_chooser 函数。query_chooser 演示了检查 SQL 表达式元素以尝试确定请求的单个分片。

构建通用分片例程是解决将实例组织在多个数据库中的问题的一种雄心勃勃的方法。对于更直接的替代方案，“不同实体”方法是一种简单的将对象分配给不同表（以及潜在的数据库节点）的显式方法 - 在维基上描述为EntityName。

文件列表：

separate_databases.py - 演示使用不同的 SQLite 数据库进行分片。
separate_tables.py - 演示使用单个 SQLite 数据库进行分片，但将使用命名约定创建多个表。
separate_schema_translates.py - 演示使用具有多个模式的单个数据库进行分片，其中可以为每个分片使用不同的“schema_translates_map”。
asyncio.py - 演示与 asyncio 一起使用分片 API。### 属性检测

演示对 SQLAlchemy 属性管理系统的修改的示例。

文件列表：

listen_for_events.py - 演示如何将事件附加到所有被检测属性并监听更改事件。
active_column_defaults.py - 演示使用AttributeEvents.init_scalar()事件，结合核心列默认值，为 ORM 对象提供在访问未设置属性时自动生成默认值的功能。
custom_management.py - 演示自定义类检测，使用sqlalchemy.ext.instrumentation扩展包。

水平分片

使用 SQLAlchemy 分片 API 的基本示例。分片是指在多个数据库之间水平扩展数据。

“分片”映射的基本组件包括：

多个Engine 实例，每个分配一个“分片 id”。这些Engine 实例可以引用不同的数据库，或者同一数据库中的不同模式/帐户，或者它们甚至可以仅通过选项进行区分，这些选项将在使用时导致它们访问不同的模式或表。
一个函数可以返回给定要保存的实例的单个分片 id；这称为“shard_chooser”。
一个函数可以返回适用于特定实例标识符的分片 id 列表；这称为“id_chooser”。如果返回所有分片 id，则将搜索所有分片。
一个函数可以返回给定查询的尝试分片 id 列表；如果返回所有分片 id，则将查询所有分片并将结果连接在一起。

在这些示例中，针对同一基本示例使用了不同类型的分片，该示例适用于按大陆基础提供天气数据。我们提供了示例 shard_chooser、id_chooser 和 query_chooser 函数。query_chooser 展示了对 SQL 表达式元素的检查，以尝试确定请求的单个分片。

构建通用分片例程是组织实例在多个数据库中的一种雄心勃勃的方法。对于更直接的替代方案，“不同实体”方法是一种将对象分配给不同表（和潜在的数据库节点）的简单方法 - 在维基上描述为EntityName。

文件列表：

separate_databases.py - 展示了使用不同的 SQLite 数据库进行分片的示例。
separate_tables.py - 展示了使用单个 SQLite 数据库进行分片的示例，但是会使用多个表并遵循命名约定。
separate_schema_translates.py - 展示了使用具有多个模式的单个数据库进行分片的示例，其中每个分片可以使用不同的“schema_translates_map”。
asyncio.py - 展示了与 asyncio 一起使用的分片 API。

扩展 ORM

ORM 查询事件

展示了增强 ORM SELECT 行为的示例，这些示例由 Session.execute() 与 2.0 风格的 select() 以及 1.x 风格的 Query 对象一起使用。

示例包括演示with_loader_criteria()选项以及SessionEvents.do_orm_execute()钩子。

截至 SQLAlchemy 1.4 版本，Query 构造与Select 构造合并在一起，因此这两个对象大部分相同。

文件列表：

temporal_range.py - 演示将应用于选定实体的自定义每个查询条件。
filter_public.py - 演示应用于特定类型实体的全局条件的基本技术。### Dogpile Caching

演示如何嵌入dogpile.cache功能与 ORM 查询，允许完全控制缓存以及从长期缓存中获取“延迟加载”属性的能力。

在这个演示中，演示了以下技术：

使用SessionEvents.do_orm_execute()事件钩子
绕过Session.execute()以从自定义缓存源而不是数据库中提取的基本技术。
使用 dogpile.cache 进行基本缓存，使用“regions”允许全局控制一组固定配置。
使用自定义UserDefinedOption对象配置语句对象中的选项。

另请参阅

重新执行语句 - 包括这里介绍的技术的一个通用示例。

例如：

# query for Person objects, specifying cache
stmt = select(Person).options(FromCache("default"))

# specify that each Person's "addresses" collection comes from
# cache too
stmt = stmt.options(RelationshipCache(Person.addresses, "default"))

# execute and results
result = session.execute(stmt)

print(result.scalars().all())

要运行，必须安装 SQLAlchemy 和 dogpile.cache，或者在当前 PYTHONPATH 上。演示将为数据文件创建一个本地目录，插入初始数据，然后运行。第二次运行演示将利用已经存在的缓存文件，并且只会发出一条 SQL 语句针对两个表 - 但显示的结果将利用数十个从缓存中获取的延迟加载。

演示脚本本身，按复杂性顺序作为 Python 模块运行，以便相对导入正常工作：

python -m examples.dogpile_caching.helloworld

python -m examples.dogpile_caching.relationship_caching

python -m examples.dogpile_caching.advanced

python -m examples.dogpile_caching.local_session_caching

文件列表：

environment.py - 建立数据/缓存文件路径和配置，必要时引导固定数据。
caching_query.py - 代表允许在 SQLAlchemy 中使用 Dogpile 缓存的函数和类。引入一个名为 FromCache 的查询选项。
model.py - 数据模型，表示具有多个地址对象的人员，每个地址对象都有邮政编码、城市和国家。
fixture_data.py - 安装一些示例数据。这里有一些美国/加拿大城市的少数邮政编码。然后，安装了 100 个人员记录，每个记录都有一个随机选择的邮政编码。
helloworld.py - 演示了如何加载一些数据，并缓存结果。
relationship_caching.py - 展示了如何在关系终点添加缓存选项，以便惰性加载从缓存中加载。
advanced.py - 演示了 Query 与 FromCache 选项的使用，包括前端加载、缓存失效和集合缓存。
local_session_caching.py - 这个例子创建了一个新的 dogpile.cache 后端，它将数据持久化存储在当前会话的字典中。删除会话，缓存就消失了。### ORM 查询事件

说明如何使用Session.execute()与 2.0 样式的select()一起增强 ORM SELECT 行为的示例，以及 1.x 样式的Query对象。

示例包括演示with_loader_criteria()选项以及SessionEvents.do_orm_execute()钩子。

从 SQLAlchemy 1.4 开始，Query构造与Select构造统一，因此这两个对象大部分相同。

文件列表：

temporal_range.py - 展示了将应用于选定实体的自定义每个查询条件。
filter_public.py - 展示了应用于特定类型实体的全局条件。

Dogpile 缓存

说明如何在 ORM 查询中嵌入dogpile.cache功能，允许完全的缓存控制，以及从长期缓存中拉取“惰性加载”属性的能力。

这个演示展示了以下技术：

使用SessionEvents.do_orm_execute()事件挂钩
绕过Session.execute()以从自定义缓存源而不是数据库中提取的基本技术。
基本的缓存与 dogpile.cache，使用允许全局控制一组固定配置的“区域”。
使用自定义UserDefinedOption对象配置语句对象中的选项。

另请参阅

重新执行语句 - 包括这里介绍的技术的一般示例。

例如：

# query for Person objects, specifying cache
stmt = select(Person).options(FromCache("default"))

# specify that each Person's "addresses" collection comes from
# cache too
stmt = stmt.options(RelationshipCache(Person.addresses, "default"))

# execute and results
result = session.execute(stmt)

print(result.scalars().all())

要运行，必须安装 SQLAlchemy 和 dogpile.cache，或者在当前 PYTHONPATH 上安装。演示将为数据文件创建一个本地目录，插入初始数据，并运行。再次运行演示将利用已经存在的缓存文件，并且只会发出一条针对两个表的 SQL 语句 - 但显示的结果将利用几十个懒加载，所有这些懒加载都从缓存中拉取。

演示脚本本身，按复杂度顺序运行为 Python 模块，以便相对导入正常工作：

python -m examples.dogpile_caching.helloworld

python -m examples.dogpile_caching.relationship_caching

python -m examples.dogpile_caching.advanced

python -m examples.dogpile_caching.local_session_caching

文件清单：

environment.py - 确定数据/缓存文件路径和配置，如有必要，引导安装装置数据。
caching_query.py - 表示使用 SQLAlchemy 可以与 Dogpile 缓存一起使用的函数和类。介绍了一个名为 FromCache 的查询选项。
model.py - 数据模型，表示具有多个 Address 对象的 Person，每个 Address 对象都有 PostalCode、City、Country。
fixture_data.py - 安装一些示例数据。这里有一些美国/加拿大几个城市的一些邮政编码。然后，安装了 100 个 Person 记录，每个记录都有一个随机选择的邮政编码。
helloworld.py - 演示如何加载一些数据，并缓存结果。
relationship_caching.py - 演示如何在关系端点上添加缓存选项，以便延迟加载从缓存加载。
advanced.py - 演示 Query 与 FromCache 选项的使用，包括前端加载、缓存失效和集合缓存。
local_session_caching.py - 此示例创建一个新的 dogpile.cache 后端，该后端将数据持久化在当前会话的字典中。删除会话，缓存消失。

标签：__,十七,sqlalchemy,self,py,hybrid,SqlAlchemy,文档,property
From： https://www.cnblogs.com/apachecn/p/18262081

SqlAlchemy-2-0-中文文档-十七-

SqlAlchemy 2.0 中文文档（十七）

排序列表

API 参考

API 参考

水平分片

API 文档

API 文档

混合属性

定义与属性行为不同的表达行为

定义设置器

允许批量 ORM 更新

处理关系

连接依赖关系混合

相关子查询关系混合

构建自定义比较器

混合值对象

API 参考

定义与属性行为不同的表达行为

使用 inplace 创建符合 pep-484 的混合属性

定义 Setter

允许批量 ORM 更新

与关系一起工作

Join-Dependent Relationship Hybrid

相关子查询关系混合

连接依赖关系混合

关联子查询关系混合

构建自定义比较器

在子类之间重用混合属性

混合值对象

API 参考

Indexable

概要

默认值

子类化

API 参考

概要

默认值

子类化

API 参考

替代类仪器化

API 参考

API 参考

ORM 示例

映射配方

邻接列表

文件列表

使用时间运行所有测试

为单个测试生成配置文件

编写您自己的套件

使用历史表进行版本控制

基本继承映射

特殊 API

属性仪器化

扩展 ORM

ORM 查询事件

Dogpile 缓存

映射配方

邻接表

文件列表

运行所有测试并计时

为个别测试转储配置文件

编写您自己的套件

使用历史表进行版本控制

关联

Asyncio 集成

有向图

动态关系作为字典

通用关联

材料化路径

嵌套集

性能

文件清单

使用时间运行所有测试

为个别测试输出简要概要

编写你自己的测试套件

文件列表

使用时间运行所有测试

为个别测试输出简要概要

编写你自己的测试套件

使用 `inplace` 创建符合 pep-484 的混合属性