SqlAlchemy 2.0 中文文档（三十三）（2）https://developer.aliyun.com/article/1562714
特殊 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 艺术在文本控制台中运行。  要运行： ```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
SqlAlchemy 2.0 中文文档（三十三）（4）https://developer.aliyun.com/article/1562718
