AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

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

2024-07-08 17
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： SqlAlchemy 2.0 中文文档（三十三）

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


动态关系作为字典

演示了如何在“动态”关系之上放置类似于字典的外观，以便字典操作（假设简单的字符串键）可以在大型集合上进行操作，而无需一次加载整个集合。

文件清单：

  • 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 艺术运行。

运行：

python -m examples.space_invaders.space_invaders

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

tail -f space_invaders.log

尽情享受！

文件清单：

  • space_invaders.py


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

文章标签：
Python
测试技术
API
数据库连接
存储
布客飞龙
目录
相关文章
布客飞龙
|
3月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（二十九）（3）
SqlAlchemy 2.0 中文文档（二十九）
布客飞龙
36 4
布客飞龙
|
3月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（二十九）（2）
SqlAlchemy 2.0 中文文档（二十九）
布客飞龙
34 7
布客飞龙
|
3月前
|
SQL 存储 关系型数据库
SqlAlchemy 2.0 中文文档（二十九）（1）
SqlAlchemy 2.0 中文文档（二十九）
布客飞龙
41 4
布客飞龙
|
3月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（三十四）（5）
SqlAlchemy 2.0 中文文档（三十四）
布客飞龙
35 0
布客飞龙
|
3月前
|
SQL 存储 关系型数据库
SqlAlchemy 2.0 中文文档（三十四）（4）
SqlAlchemy 2.0 中文文档（三十四）
布客飞龙
37 1
布客飞龙
|
3月前
|
SQL 数据库 Python
SqlAlchemy 2.0 中文文档（三十一）（3）
SqlAlchemy 2.0 中文文档（三十一）
布客飞龙
23 1
布客飞龙
|
3月前
|
存储 SQL 测试技术
SqlAlchemy 2.0 中文文档（三十一）（1）
SqlAlchemy 2.0 中文文档（三十一）
布客飞龙
29 1
布客飞龙
|
3月前
|
JSON 测试技术 数据格式
SqlAlchemy 2.0 中文文档（三十一）（4）
SqlAlchemy 2.0 中文文档（三十一）
布客飞龙
31 1
布客飞龙
|
3月前
|
SQL 测试技术 数据库
SqlAlchemy 2.0 中文文档（三十一）（2）
SqlAlchemy 2.0 中文文档（三十一）
布客飞龙
24 1
布客飞龙
|
3月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（二十九）（4）
SqlAlchemy 2.0 中文文档（二十九）
布客飞龙
35 4