SqlAlchemy 2.0 中文文档（五十七）（2）

SqlAlchemy 2.0 中文文档（五十七）（1）https://developer.aliyun.com/article/1563163

基准测试

SQLAlchemy 在examples/目录中包含一个性能套件，我们可以利用bulk_insert套件以不同的方式使用 Core 和 ORM 来对许多行进行 INSERT 的基准测试。

对于以下测试，我们正在插入100,000 个对象，在所有情况下，我们实际上在内存中有 100,000 个真实的 Python ORM 对象，要么是预先创建的，要么是动态生成的。除了 SQLite 之外的所有数据库都通过本地网络连接运行，而不是 localhost；这导致“较慢”的结果非常慢。

通过此功能改进的操作包括：

• 使用Session.add()和Session.add_all()将对象添加到会话中的工作单元刷新。
  
• 新的 ORM 批量插入语句功能，改进了 SQLAlchemy 1.4 中首次引入的实验性版本。
  
• 在 Bulk Operations 中描述的Session“批量”操作，被上述 ORM 批量插入功能取代。
  

为了了解操作的规模，以下是使用test_flush_no_pk性能套件进行的性能测量，该套件历史上代表了  SQLAlchemy 的最坏情况 INSERT 性能任务，其中需要 INSERT  没有主键值的对象，然后必须获取新生成的主键值，以便这些对象可以用于后续的 flush 操作，比如在关系中建立关系，刷新联合继承模型等：

@Profiler.profile
def test_flush_no_pk(n):
  """INSERT statements via the ORM (batched with RETURNING if available),
 fetching generated row id"""
    session = Session(bind=engine)
    for chunk in range(0, n, 1000):
        session.add_all(
            [
                Customer(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
                for i in range(chunk, chunk + 1000)
            ]
        )
        session.flush()
    session.commit()

可以从任何 SQLAlchemy 源代码树中运行此测试：

python -m examples.performance.bulk_inserts --test test_flush_no_pk

下表总结了最新的 1.4 系列 SQLAlchemy 与 2.0 在运行相同测试时的性能测量：

注意

另外两个驱动程序在性能上没有变化；psycopg2 驱动程序，在 SQLAlchemy 1.4 中已经实现了快速 executemany，以及 MySQL，继续不提供 RETURNING 支持：

变更摘要

下面的项目列出了 2.0 中为使所有驱动程序达到这种状态而进行的各个更改：

• SQLite 实现了 RETURNING - #6195
  
• MariaDB 实现了 RETURNING - #7011
  
• 修复 Oracle 的多行 RETURNING - #6245
  
• 使 insert() executemany()支持尽可能多的方言，通常使用 VALUES() - #6047
  
• 当 RETURNING 与 executemany 一起用于不支持的后端时发出警告（当前没有 RETURNING 后端有此限制） - #7907
  
• ORM Mapper.eager_defaults 参数现在默认为新设置"auto"，当使用的后端支持“insertmanyvalues”时，将自动为 INSERT 语句启用“急切默认值”。请参阅获取服务器生成的默认值以获取文档。
  

另请参阅

“插入多个值”行为适用于 INSERT 语句 - 新功能的文档和背景以及如何配置它的说明 ## 启用 ORM 的插入、更新和删除语句，带有 ORM RETURNING

SQLAlchemy 1.4 将传统的Query对象的特性移植到 2.0 风格的执行中，这意味着Select构造可以传递给Session.execute()以提供 ORM 结果。还添加了对Update和Delete的支持，可以传递给Session.execute()，以便它们可以提供Query.update()和Query.delete()的实现。

主要缺失的元素是对Insert构造的支持。1.4 文档通过一些关于在 ORM 上下文中使用Select.from_statement()来集成 RETURNING 的“插入”和“upserts”的示例来解决这个问题。2.0 现在通过将Insert直接支持作为Session.bulk_insert_mappings()方法的增强版本，以及对所有 DML 结构的完整 ORM RETURNING 支持来完全弥补这一差距。

带有 RETURNING 的批量插入

Insert可以传递给Session.execute()，可以带有或不带有Insert.returning()，当与一个单独的参数列表一起传递时，将调用与以前由Session.bulk_insert_mappings()实现的相同过程，同时还添加了额外的增强功能。这将通过利用新的快速插入多行功能来优化行的批处理，同时还支持异构参数集和多表映射，如联合表继承：

>>> users = session.scalars(
...     insert(User).returning(User),
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ],
... )
>>> print(users.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
 User(name='sandy', fullname='Sandy Cheeks'),
 User(name='patrick', fullname='Patrick Star'),
 User(name='squidward', fullname='Squidward Tentacles'),
 User(name='ehkrabs', fullname='Eugene H. Krabs')]

RETURNING 支持所有这些用例，其中 ORM 将从多个语句调用构造完整的结果集。

另请参见

ORM 批量 INSERT 语句

批量 UPDATE

与Insert类似，将Update构造与包含主键值的参数列表一起传递给Session.execute()将调用与以前由Session.bulk_update_mappings()方法支持的相同过程。但是，此功能不支持 RETURNING，因为它使用 SQL UPDATE 语句，该语句使用 DBAPI executemany��行调用：

>>> from sqlalchemy import update
>>> session.execute(
...     update(User),
...     [
...         {"id": 1, "fullname": "Spongebob Squarepants"},
...         {"id": 3, "fullname": "Patrick Star"},
...     ],
... )

另请参见

按主键进行 ORM 批量 UPDATE

INSERT / upsert … VALUES … RETURNING

当使用Insert与Insert.values()时，参数集合可能包含 SQL 表达式。此外，还支持 SQLite、PostgreSQL 和 MariaDB 等数据库的 upsert 变体。这些语句现在可以包括带有列表达式或完整 ORM 实体的Insert.returning()子句：

>>> from sqlalchemy.dialects.sqlite import insert as sqlite_upsert
>>> stmt = sqlite_upsert(User).values(
...     [
...         {"name": "spongebob", "fullname": "Spongebob Squarepants"},
...         {"name": "sandy", "fullname": "Sandy Cheeks"},
...         {"name": "patrick", "fullname": "Patrick Star"},
...         {"name": "squidward", "fullname": "Squidward Tentacles"},
...         {"name": "ehkrabs", "fullname": "Eugene H. Krabs"},
...     ]
... )
>>> stmt = stmt.on_conflict_do_update(
...     index_elements=[User.name], set_=dict(fullname=stmt.excluded.fullname)
... )
>>> result = session.scalars(stmt.returning(User))
>>> print(result.all())
[User(name='spongebob', fullname='Spongebob Squarepants'),
User(name='sandy', fullname='Sandy Cheeks'),
User(name='patrick', fullname='Patrick Star'),
User(name='squidward', fullname='Squidward Tentacles'),
User(name='ehkrabs', fullname='Eugene H. Krabs')]

另请参见

ORM 批量插入带有每行 SQL 表达式

ORM “upsert”语句

带有 WHERE … RETURNING 的 ORM UPDATE / DELETE

SQLAlchemy 1.4 还对 RETURNING 功能提供了一些支持，可与update()和delete()构造一起使用，当与Session.execute()一起使用时。此支持现已升级为完全本地化，包括fetch同步策略是否存在 RETURNING 的明确使用：

>>> from sqlalchemy import update
>>> stmt = (
...     update(User)
...     .where(User.name == "squidward")
...     .values(name="spongebob")
...     .returning(User)
... )
>>> result = session.scalars(stmt, execution_options={"synchronize_session": "fetch"})
>>> print(result.all())

另请参见

带有自定义 WHERE 条件的 ORM UPDATE 和 DELETE

使用 RETURNING 进行 UPDATE/DELETE 和自定义 WHERE 条件

改进了 ORM UPDATE / DELETE 的synchronize_session行为

synchronize_session 的默认策略现在是一个新值"auto"。此策略将尝试使用"evaluate"策略，然后自动回退到"fetch"策略。对于除了 MySQL / MariaDB 之外的所有后端，"fetch"使用 RETURNING 在同一语句中获取已更新/删除的主键标识符，因此通常比以前版本更有效（在 1.4 版本中，RETURNING 仅适用于 PostgreSQL、SQL Server）。

亦见

选择同步策略

变更摘要

新的 ORM DML 带有 RETURNING 特性的已列出的票证：

• 将 ORM 级别的insert()转换为在 ORM 上下文中解释values() - #7864
  
• 评估dml.returning(Entity)的可行性，以提供 ORM 表达式，自动应用select().from_statement等效 - #7865
  
• 给定 ORM 插入，尝试携带批量方法，有关继承 - #8360 ## 新的“仅写入”关系策略取代了“动态”
  

“懒加载”策略lazy="dynamic"已经过时，因为它被硬编码为使用传统的Query。这个加载策略既不兼容 asyncio，而且还有许多行为隐式迭代其内容，这违背了“动态”关系的原始目的，即用于非常大的集合，不应随时隐式加载到内存中。

“动态”策略现已由新策略lazy="write_only"取代。可以使用relationship.lazy参数的配置relationship()来实现“仅写入”，或者在使用类型注释映射时，指示WriteOnlyMapped注解作为映射样式：

from sqlalchemy.orm import WriteOnlyMapped
class Base(DeclarativeBase):
    pass
class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: WriteOnlyMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )
class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

写入唯一映射集合类似于lazy="dynamic"，因为集合可以提前分配，并且还具有诸如WriteOnlyCollection.add()和WriteOnlyCollection.remove()等方法，以逐个项目的方式修改集合：

new_account = Account(
    identifier="account_01",
    account_transactions=[
        AccountTransaction(description="initial deposit", amount=Decimal("500.00")),
        AccountTransaction(description="transfer", amount=Decimal("1000.00")),
        AccountTransaction(description="withdrawal", amount=Decimal("-29.50")),
    ],
)
new_account.account_transactions.add(
    AccountTransaction(description="transfer", amount=Decimal("2000.00"))
)

更大的区别在于数据库加载方面，集合无法直接从数据库加载对象；而是使用诸如 WriteOnlyCollection.select() 等 SQL 构造方法来生成诸如 Select 的 SQL 构造，然后使用 2.0 风格 以显式方式加载所需对象：

account_transactions = session.scalars(
    existing_account.account_transactions.select()
    .where(AccountTransaction.amount < 0)
    .limit(10)
).all()

WriteOnlyCollection 也与新的 ORM 批量 DML 特性集成，包括支持带有 WHERE 条件的批量 INSERT 和 UPDATE/DELETE，全部支持 RETURNING。详见完整文档 Write Only Relationships。

参见

Write Only Relationships

为动态关系添加了新的 pep-484 / 类型注释映射支持

尽管“动态”关系在 2.0 中是遗留的，但由于这些模式预计具有很长的寿命，类型注释映射 现在也为“动态”关系添加了支持，方式与新的 lazy="write_only" 方法相同，使用 DynamicMapped 注释：

from sqlalchemy.orm import DynamicMapped
class Base(DeclarativeBase):
    pass
class Account(Base):
    __tablename__ = "account"
    id: Mapped[int] = mapped_column(primary_key=True)
    identifier: Mapped[str]
    account_transactions: DynamicMapped["AccountTransaction"] = relationship(
        cascade="all, delete-orphan",
        passive_deletes=True,
        order_by="AccountTransaction.timestamp",
    )
class AccountTransaction(Base):
    __tablename__ = "account_transaction"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(
        ForeignKey("account.id", ondelete="cascade")
    )
    description: Mapped[str]
    amount: Mapped[Decimal]
    timestamp: Mapped[datetime] = mapped_column(default=func.now())

上述映射将提供一个类型为 AppenderQuery 集合类型的 Account.account_transactions 集合，包括其元素类型，例如 AppenderQuery[AccountTransaction]。然后允许迭代和查询产生类型为 AccountTransaction 的对象。

参见

动态关系加载器

#7123 ## 安装现在完全启用了 pep-517

源发行版现在包括一个 pyproject.toml 文件，以支持完整的 PEP 517 支持。特别是，这允许使用 pip 进行本地源构建时自动安装 Cython 可选依赖项。

#7311 ## C 扩展现在已转换为 Cython

SQLAlchemy C 扩展已被全部采用 Cython 编写的全新扩展所取代。虽然在创建 C 扩展时曾于 2010 年评估过 Cython，但如今实际使用的 C 扩展的性质和重点与当时已经发生了很大变化。与此同时，Cython 显然也有了显著的发展，Python 的构建/分发工具链也使我们有可能重新审视它。

切换到 Cython 提供了明显的新优势，而没有明显的不利因素：

• 替换特定 C 扩展的 Cython 扩展已经被全部作为更快的扩展进行了基准测试，通常情况下稍微快一点，但有时比  SQLAlchemy 以前包含的几乎所有 C 代码都要显著快。虽然这看起来很神奇，但似乎是 Cython  实现中的一些非显而易见的优化的产物，在许多情况下，这些优化不会出现在直接的 Python 到 C 的函数移植中，特别是对于许多添加到 C  扩展中的自定义集合类型而言。
  
• 与原始的 C 代码相比，Cython 扩展编写、维护和调试都要容易得多，在大多数情况下与 Python 代码几乎一致。预计未来的版本中会有更多的 SQLAlchemy 元素被移植到 Cython 中，这将打开许多以前无法触及的性能改进的新门。
  
• Cython 非常成熟并被广泛使用，包括成为 SQLAlchemy 支持的一些著名数据库驱动程序的基础，包括 asyncpg、psycopg3 和 asyncmy。
  

与之前的 C 扩展一样，Cython 扩展已经预先构建在 SQLAlchemy 的 wheel 发布中，这些发布会自动提供给 pip 从 PyPi 安装。手动构建说明也没有改变，除了对 Cython 的要求。

另请参阅

构建 Cython 扩展

#7256 ## 数据库反射的主要架构、性能和 API 增强

完全重新设计了 Table 对象及其组件 反射 的内部系统，以允许参与的方言一次性高性能地批量反射数千个表。目前，PostgreSQL 和 Oracle 方言参与了新的架构，其中 PostgreSQL 方言现在可以比以前快近三倍地反射大量的 Table 对象，而 Oracle 方言现在可以比以前快十倍地反射大量的 Table 对象。

重新架构主要适用于使用 SELECT 查询系统目录表以反映表的方言，而其余包含的方言可以从这种方法中受益的是 SQL Server  方言。相比之下，MySQL/MariaDB 和 SQLite 方言使用非关系型系统来反映数据库表，并且没有受到现有性能问题的影响。

新 API 兼容先前的系统，并且不需要更改第三方方言以保持兼容性；第三方方言也可以通过实现批量查询模式反射模式来选择新系统。

除此之外，Inspector 对象的 API 和行为已经改进和增强，具有更一致的跨方言行为以及新方法和新性能特性。

性能概述

源分发包括一个脚本 test/perf/many_table_reflection.py，用于测试现有的反射功能以及新功能。其中一部分测试可以在旧版本的 SQLAlchemy 上运行，我们在这里使用它来说明性能差异，以在本地网络连接上一次性反射 250 个 Table 对象：

Inspector() 的行为变更

对于 SQLAlchemy 包含的 SQLite、PostgreSQL、MySQL/MariaDB、Oracle 和 SQL Server 的方言，Inspector.has_table()、Inspector.has_sequence()、Inspector.has_index()、Inspector.get_table_names()和Inspector.get_sequence_names()现在在缓存方面都表现一致：在为特定Inspector对象第一次调用后，它们都会完全缓存其结果。在调用相同的Inspector对象时创建或删除表/序列的程序将在数据库状态更改后不会收到更新的状态。当要执行 DDL 更改时，应使用调用Inspector.clear_cache()或新的Inspector。以前，Inspector.has_table()、Inspector.has_sequence()方法未实现缓存，Inspector也不支持这些方法的缓存，而Inspector.get_table_names()和Inspector.get_sequence_names()方法则是，导致两种方法之间的结果不一致。

第三方方言的行为取决于它们是否实现了这些方法的方言级别实现的“反射缓存”装饰器。

新的方法和Inspector()的改进

• 添加了一个方法Inspector.has_schema()，用于返回目标数据库中是否存在模式
  
• 添加了一个方法Inspector.has_index()，用于返回表是否具有特定索引。
  
• 对一次只对单个表起作用的检查方法，例如Inspector.get_columns()，现在应该一致地引发NoSuchTableError如果找不到表或视图; 此更改特定于各个方言，因此对于现有的第三方方言可能不适用。
  
• 将“视图”和“材料化视图”的处理分开，因为在现实世界的用例中，这两个构造使用不同的 DDL 来创建和删除; 这包括现在有单独的Inspector.get_view_names()和Inspector.get_materialized_view_names()方法。
  

#4379 ## 对 psycopg 3（也称为“psycopg”）的方言支持

为psycopg 3 DBAPI 增加了方言支持，尽管现在被称为psycopg，但它的包名仍然取代了之前的psycopg2包，后者目前仍然是 SQLAlchemy“默认”的postgresql方言驱动程序。 psycopg是一个完全重新设计和现代化的用于 PostgreSQL 的数据库适配器，支持诸如准备语句和 Python asyncio 等概念。

psycopg是 SQLAlchemy 支持的第一个同时提供 pep-249 同步 API 和 asyncio 驱动程序的 DBAPI。可以使用相同的psycopg数据库 URL 与create_engine()和create_async_engine()引擎创建函数，自动选择相应的同步或 asyncio 版本的方言。

另请参阅

psycopg ## 对 oracledb 的方言支持

为oracledb DBAPI 增加了方言支持，这是流行的 cx_Oracle 驱动程序的重命名的新主要版本。

另请参阅

python-oracledb ## 新的条件 DDL 用于约束和索引

一个新的方法 Constraint.ddl_if() 和 Index.ddl_if() 允许诸如 CheckConstraint、UniqueConstraint 和 Index 这样的构造在给定的 Table 上有条件地呈现，基于与 DDLElement.execute_if() 方法接受的相同类型的条件。在下面的示例中，CHECK 约束和索引只会针对 PostgreSQL 后端生成：

meta = MetaData()
my_table = Table(
    "my_table",
    meta,
    Column("id", Integer, primary_key=True),
    Column("num", Integer),
    Column("data", String),
    Index("my_pg_index", "data").ddl_if(dialect="postgresql"),
    CheckConstraint("num > 5").ddl_if(dialect="postgresql"),
)
e1 = create_engine("sqlite://", echo=True)
meta.create_all(e1)  # will not generate CHECK and INDEX
e2 = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
meta.create_all(e2)  # will generate CHECK and INDEX

另见

控制约束和索引的 DDL 生成

#7631 ## DATE、TIME、DATETIME 数据类型现在在所有后端上都支持字面渲染

字面渲染现在已经实现了对于日期和时间类型的后端特定编译，包括 PostgreSQL 和 Oracle：

>>> import datetime
>>> from sqlalchemy import DATETIME
>>> from sqlalchemy import literal
>>> from sqlalchemy.dialects import oracle
>>> from sqlalchemy.dialects import postgresql
>>> date_literal = literal(datetime.datetime.now(), DATETIME)
>>> print(
...     date_literal.compile(
...         dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
'2022-12-17 11:02:13.575789'
>>> print(
...     date_literal.compile(
...         dialect=oracle.dialect(), compile_kwargs={"literal_binds": True}
...     )
... )
TO_TIMESTAMP('2022-12-17 11:02:13.575789',  'YYYY-MM-DD HH24:MI:SS.FF') 

以前，这样的字面渲染仅在未提供任何方言的情况下将语句转换为字符串时才起作用；当尝试使用特定于方言的类型进行渲染时，会引发 NotImplementedError，直到 SQLAlchemy 1.4.45，这变成了一个 CompileError（部分来源于 #8800）。

当使用 PostgreSQL、MySQL、MariaDB、MSSQL、Oracle 方言提供的 SQL 编译器的 literal_binds 时，默认渲染是修改后的 ISO-8601 渲染（即将 T 转换为空格的 ISO-8601）。对于 Oracle，ISO 格式被包装在适当的 TO_DATE() 函数调用中。对于 SQLite，渲染保持不变，因为这个方言始终为日期值包含字符串渲染。

#5052 ## Result、AsyncResult 的上下文管理器支持

Result 对象现在支持上下文管理器的使用，这将确保对象及其底层游标在块结束时关闭。这在特定于服务器端游标的情况下特别有用，在这种情况下，重要的是在操作结束时关闭打开的游标对象，即使发生了用户定义的异常：

with engine.connect() as conn:
    with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

在使用 asyncio 时，AsyncResult和AsyncConnection已经改变，以提供可选的异步上下文管理器使用，如下所示：

async with async_engine.connect() as conn:
    async with conn.execution_options(yield_per=100).execute(
        text("select * from table")
    ) as result:
        for row in result:
            print(f"{row}")

#8710

SqlAlchemy 2.0 中文文档（五十七）（3）https://developer.aliyun.com/article/1563165