SQLAlchemy 2.0 的新特性？

读者须知

SQLAlchemy 2.0 的过渡文档分为 两 个文档 - 一个详细介绍了从 1.x 到 2.x 系列的主要 API 转变，另一个详细介绍了相对于 SQLAlchemy 1.4 的新特性和行为

尚未将其 1.4 应用程序更新为遵循 SQLAlchemy 2.0 引擎和 ORM 约定的读者，可以导航到 SQLAlchemy 2.0 - 主要迁移指南，以获取确保 SQLAlchemy 2.0 兼容性的指南，这是在版本 2.0 下运行工作代码的先决条件。

关于本文档

本文档描述了 SQLAlchemy 版本 1.4 和 SQLAlchemy 版本 2.0 之间的更改，独立于 1.x 风格 和 2.0 风格 用法之间的主要更改。读者应从 SQLAlchemy 2.0 - 主要迁移指南 文档开始，以全面了解 1.x 和 2.x 系列之间的主要兼容性更改。

除了主要的 1.x->2.x 迁移路径外，SQLAlchemy 2.0 中下一个最大的范式转变是与 PEP 484 类型实践和当前功能的深度集成，尤其是在 ORM 中。受 Python dataclasses 启发的新型类型驱动的 ORM 声明式风格，以及与 dataclass 本身的新集成，补充了一种整体方法，该方法不再需要 stubs，并且在从 SQL 语句到结果集提供类型感知方法链方面取得了很大进展。

Python 类型化的突出地位不仅在于使诸如 mypy 之类的类型检查器可以在没有插件的情况下运行；更重要的是，它使诸如 vscode 和 pycharm 之类的 IDE 可以在协助构建 SQLAlchemy 应用程序方面发挥更积极的作用。

核心和 ORM 中的新类型支持 - 不再使用 Stubs / 扩展

与版本 1.4 中通过 sqlalchemy2-stubs 包提供的临时方法相比，核心和 ORM 的类型化方法已完全重新设计。新方法从 SQLAlchemy 中最基本的元素开始，即 Column，或者更准确地说是 ColumnElement，它是所有具有类型的 SQL 表达式的基础。这种表达式级别的类型化然后扩展到语句构造、语句执行和结果集领域，最后扩展到 ORM，在 ORM 中，新的 声明式 形式允许完全类型化的 ORM 模型，该模型从语句一直集成到结果集。

提示

对于 2.0 系列，类型支持应被视为 beta 级别 的软件。类型化细节可能会发生变化，但未计划进行重大的向后不兼容的更改。

SQL 表达式 / 语句 / 结果集 类型

本节提供了 SQLAlchemy 新的 SQL 表达式类型化方法的背景和示例，该方法从基本 ColumnElement 构造扩展到 SQL 语句和结果集，并进入 ORM 映射领域。

理由和概述

提示

本节是架构讨论。跳到 SQL 表达式类型 - 示例 以查看新类型化的外观。

在 sqlalchemy2-stubs 中，SQL 表达式被类型化为 泛型，然后引用 TypeEngine 对象，例如 Integer、DateTime 或 String 作为其泛型参数（例如 Column[Integer]）。这本身就与最初的 Dropbox sqlalchemy-stubs 包的做法有所不同，在 Dropbox 包中，Column 及其基础构造直接是 Python 类型的泛型，例如 int、datetime 和 str。人们希望，由于 Integer / DateTime / String 本身是针对 int / datetime / str 的泛型，因此可以维护两个级别的信息，并且能够通过 TypeEngine 作为中间构造从列表达式中提取 Python 类型。然而，事实并非如此，因为 PEP 484 实际上并没有足够丰富的功能集使其可行，它缺少诸如 更高种类 TypeVars 之类的功能。

因此，在对 PEP 484 的当前功能进行 深入评估 之后，SQLAlchemy 2.0 认识到了 sqlalchemy-stubs 在此领域的原始智慧，并恢复为将列表达式直接链接到 Python 类型。这确实意味着，如果一个人的 SQL 表达式是不同的子类型，例如 Column(VARCHAR) 与 Column(Unicode)，则这两个 String 子类型的具体细节不会作为类型一起传递，因为类型仅携带 str，但实际上这通常不是问题，并且 Python 类型立即存在通常会更有用，因为它表示人们将直接为此列存储和接收的 Python 数据。

具体而言，这意味着诸如 Column('id', Integer) 之类的表达式被类型化为 Column[int]。这允许建立 SQLAlchemy 构造 -> Python 数据类型的可行管道，而无需类型化插件。至关重要的是，它允许与 ORM 使用 select() 和 Row 构造的范例完全互操作，这些构造引用 ORM 映射的类类型（例如，包含用户映射实例的 Row，例如我们的教程中使用的 User 和 Address 示例）。虽然 Python 类型化目前对元组类型的自定义支持非常有限（其中 PEP 646，第一个尝试处理类似元组的对象的 pep，在 其功能上被有意限制，并且本身对于任意元组操作尚不可行），但已经设计出一种相当不错的方法，该方法允许基本的 select() -> Result -> Row 类型化起作用，包括对于 ORM 类，在 Row 对象要解包到各个列条目时，添加了一个小的面向类型化的访问器，该访问器允许各个 Python 值保持与其来源的 SQL 表达式链接的 Python 类型（翻译：它有效）。

SQL 表达式类型 - 示例

类型化行为的简要浏览。注释指示在 vscode 中将鼠标悬停在代码上时会看到的内容（或类型化工具在使用 reveal_type() 助手时大致显示的内容）

  • 分配给 SQL 表达式的简单 Python 类型

    # (variable) str_col: ColumnClause[str]
str_col = column("a", String)

# (variable) int_col: ColumnClause[int]
int_col = column("a", Integer)

# (variable) expr1: ColumnElement[str]
expr1 = str_col + "x"

# (variable) expr2: ColumnElement[int]
expr2 = int_col + 10

# (variable) expr3: ColumnElement[bool]
expr3 = int_col == 15

  • 分配给 select() 构造的各个 SQL 表达式，以及任何返回行的构造，包括返回行的 DML，例如带有 Insert 和 Insert.returning() 的 Insert，都打包到 Tuple[] 类型中，该类型保留每个元素的 Python 类型。

    # (variable) stmt: Select[Tuple[str, int]]
stmt = select(str_col, int_col)

# (variable) stmt: ReturningInsert[Tuple[str, int]]
ins_stmt = insert(table("t")).returning(str_col, int_col)

  • 来自任何返回行构造的 Tuple[] 类型，当使用 .execute() 方法调用时，会传递到 Result 和 Row。为了将 Row 对象解包为元组，Row.tuple() 或 Row.t 访问器本质上将 Row 转换为相应的 Tuple[] (尽管在运行时仍然是相同的 Row 对象)。

    with engine.connect() as conn:
    # (variable) stmt: Select[Tuple[str, int]]
    stmt = select(str_col, int_col)

    # (variable) result: Result[Tuple[str, int]]
    result = conn.execute(stmt)

    # (variable) row: Row[Tuple[str, int]] | None
    row = result.first()

    if row is not None:
        # for typed tuple unpacking or indexed access,
        # use row.tuple() or row.t  (this is the small typing-oriented accessor)
        strval, intval = row.t

        # (variable) strval: str
        strval

        # (variable) intval: int
        intval

  • 单列语句的标量值通过诸如 Connection.scalar(), Result.scalars() 等方法正确处理。

    # (variable) data: Sequence[str]
data = connection.execute(select(str_col)).scalars().all()

  • 上述对返回行构造的支持最适用于 ORM 映射类，因为映射类可以列出其成员的特定类型。下面的示例使用 新的类型感知语法 设置一个类，这将在以下部分中描述

    from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    addresses: Mapped[List["Address"]] = relationship()


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))

    通过上述映射，属性被类型化，并从语句一直表达自己到结果集

    with Session(engine) as session:
    # (variable) stmt: Select[Tuple[int, str]]
    stmt_1 = select(User.id, User.name)

    # (variable) result_1: Result[Tuple[int, str]]
    result_1 = session.execute(stmt_1)

    # (variable) intval: int
    # (variable) strval: str
    intval, strval = result_1.one().t

    映射类本身也是类型，并且行为相同，例如针对两个映射类进行 SELECT

    with Session(engine) as session:
    # (variable) stmt: Select[Tuple[User, Address]]
    stmt_2 = select(User, Address).join_from(User, Address)

    # (variable) result_2: Result[Tuple[User, Address]]
    result_2 = session.execute(stmt_2)

    # (variable) user_obj: User
    # (variable) address_obj: Address
    user_obj, address_obj = result_2.one().t

    在选择映射类时，诸如 aliased 之类的构造也有效，既保持了原始映射类的列级属性，也保持了语句预期的返回类型

    with Session(engine) as session:
    # this is in fact an Annotated type, but typing tools don't
    # generally display this

    # (variable) u1: Type[User]
    u1 = aliased(User)

    # (variable) stmt: Select[Tuple[User, User, str]]
    stmt = select(User, u1, User.name).filter(User.id == 5)

    # (variable) result: Result[Tuple[User, User, str]]
    result = session.execute(stmt)

  • 核心 Table 尚没有一种体面的方法来维护通过 Table.c 访问器访问 Column 对象时的类型化。

    由于 Table 设置为类的实例，并且 Table.c 访问器通常按名称动态访问 Column 对象，因此尚没有针对此的已建立的类型化方法；需要一些替代语法。

  • ORM 类、标量等效果很好。

    选择 ORM 类的典型用例，作为标量或元组，全部有效，包括 2.0 和 1.x 风格的查询，都能够返回精确的类型本身或包含在适当的容器中，例如 Sequence[]、List[] 或 Iterator[]

    # (variable) users1: Sequence[User]
users1 = session.scalars(select(User)).all()

# (variable) user: User
user = session.query(User).one()

# (variable) user_iter: Iterator[User]
user_iter = iter(session.scalars(select(User)))

  • 旧版 Query 也获得了元组类型化。

    Query 的类型支持远远超出了 sqlalchemy-stubs 或 sqlalchemy2-stubs 提供的功能，其中标量对象以及元组类型化的 Query 对象将在大多数情况下保留结果级别的类型化

    # (variable) q1: RowReturningQuery[Tuple[int, str]]
q1 = session.query(User.id, User.name)

# (variable) rows: List[Row[Tuple[int, str]]]
rows = q1.all()

# (variable) q2: Query[User]
q2 = session.query(User)

# (variable) users: List[User]
users = q2.all()

注意事项 - 必须卸载所有 catch-all stubs

类型支持的一个关键注意事项是，必须卸载所有 SQLAlchemy stubs 包才能使类型化正常工作。当针对 Python 虚拟环境运行 mypy 时，这只是卸载这些包的问题。但是，SQLAlchemy stubs 包目前也是 typeshed 的一部分，而 typeshed 本身已捆绑到某些类型化工具中，例如 Pylance，因此在某些情况下，可能需要找到这些包的文件并删除它们，如果它们实际上干扰了新类型化的正常工作。

一旦 SQLAlchemy 2.0 以最终状态发布，typeshed 将从其自身的 stubs 源代码中删除 SQLAlchemy。

ORM 声明式模型

SQLAlchemy 1.4 引入了第一个 SQLAlchemy 原生 ORM 类型支持，它结合使用了 sqlalchemy2-stubs 和 Mypy 插件。在 SQLAlchemy 2.0 中，Mypy 插件 仍然可用，并且已更新以与 SQLAlchemy 2.0 的类型系统一起工作。但是，它现在应被视为 已弃用，因为应用程序现在拥有采用新类型支持的直接途径，而无需使用插件或 stubs。

概述

新系统的基本方法是，当使用完全 声明式 模型（即，不是 混合声明式 或 命令式 配置，这些配置保持不变）时，映射的列声明首先在运行时通过检查每个属性声明左侧的类型注解（如果存在）来派生。左侧类型注解应包含在 Mapped 泛型类型中，否则该属性不被视为映射属性。然后，属性声明可以引用右侧的 mapped_column() 构造，该构造用于提供有关要生成和映射的 Column 的其他核心级模式信息。如果在左侧存在 Mapped 注解，则右侧声明是可选的；如果在左侧不存在注解，则可以将 mapped_column() 用作 Column 指令的精确替代，它将为属性提供更准确（但不完全精确）的类型化行为，即使不存在注解也是如此。

这种方法的灵感来源于 Python dataclasses 的方法，它以左侧的注解开始，然后允许右侧有一个可选的 dataclasses.field() 规范；与 dataclasses 方法的关键区别在于，SQLAlchemy 的方法是严格的选择性加入，其中使用 Column 而没有任何类型注解的现有映射继续像往常一样工作，并且 mapped_column() 构造可以作为 Column 的直接替代品使用，而无需任何显式的类型注解。只有当需要精确的属性级 Python 类型时，才需要使用带有 Mapped 的显式注解。这些注解可以根据需要，在每个属性的基础上用于那些特定类型有帮助的属性；使用 mapped_column() 的非注解属性将在实例级别被类型化为 Any。

迁移现有映射

过渡到新的 ORM 方法开始时会比较冗长，但随着可用新功能的充分使用，它会比以前更简洁。以下步骤详细介绍了一个典型的过渡过程，然后继续说明更多选项。

第一步 - declarative_base() 被 DeclarativeBase 取代。

Python 类型注解中观察到的一个限制是，似乎无法从函数动态生成一个类，然后该类被类型检查工具理解为新类的基类。为了在不使用插件的情况下解决这个问题，通常对 declarative_base() 的调用可以替换为使用 DeclarativeBase 类，它产生与往常相同的 Base 对象，只是类型检查工具能够理解它。

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass
第二步 - 将声明式 Column 的使用替换为 mapped_column()

mapped_column() 是一个 ORM 类型感知的构造，可以直接替换为 Column 的使用。给定一个 1.x 风格的映射如下：

from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

我们将 Column 替换为 mapped_column()；不需要更改任何参数。

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30), nullable=False)
    fullname = mapped_column(String)
    addresses = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

上面的单个列尚未用 Python 类型进行类型化，而是被类型化为 Mapped[Any]；这是因为我们可以用 Optional 或不用 Optional 来声明任何列，并且没有一种“猜测”方式可以在我们显式地对其进行类型化时不会导致类型错误。

但是，在此步骤中，我们上面的映射为所有属性设置了适当的 描述符 类型，并且可以在查询以及实例级操作中使用，所有这些都将通过 mypy –strict 模式，而无需任何插件。

第三步 - 根据需要使用 Mapped 应用精确的 Python 类型。

这可以为所有需要精确类型化的属性完成；可以跳过那些可以保留为 Any 的属性。为了提供上下文，我们还说明了 Mapped 如何用于 relationship()，我们在其中应用了精确的类型。此中间步骤中的映射将更加冗长，但是随着熟练程度的提高，此步骤可以与后续步骤结合使用，以更直接地更新映射。

from typing import List
from typing import Optional
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 User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30), nullable=False)
    fullname: Mapped[Optional[str]] = mapped_column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email_address: Mapped[str] = mapped_column(String, nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"), nullable=False)
    user: Mapped["User"] = relationship("User", back_populates="addresses")

此时，我们的 ORM 映射已完全类型化，并将生成精确类型化的 select()、Query 和 Result 构造。我们现在可以继续减少映射声明中的冗余。

第四步 - 删除不再需要的 mapped_column() 指令

所有 nullable 参数都可以使用 Optional[] 来暗示；在缺少 Optional[] 的情况下，nullable 默认为 False。所有没有参数的 SQL 类型，例如 Integer 和 String，都可以仅用 Python 注解来表示。mapped_column() 指令如果没有任何参数，则可以完全删除。relationship() 现在从左侧的注解中派生出它的类，也支持前向引用（因为 relationship() 已经支持基于字符串的前向引用十年了 😉）。

from typing import List
from typing import Optional
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 User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
第五步 - 利用 pep-593 Annotated 将常用指令打包到类型中

这是一个全新的功能，它提供了一种替代或补充 声明式混入 的方法，作为提供面向类型的配置的手段，并且在大多数情况下取代了对 declared_attr 装饰函数的需要。

首先，声明式映射允许自定义 Python 类型到 SQL 类型的映射，例如 str 到 String，使用 registry.type_annotation_map。使用 PEP 593 Annotated 允许我们创建特定 Python 类型的变体，以便可以使用相同的类型，例如 str，每个变体都提供 String 的变体，如下所示，其中使用名为 str50 的 Annotated str 将指示 String(50)。

from typing_extensions import Annotated
from sqlalchemy.orm import DeclarativeBase

str50 = Annotated[str, 50]


# declarative base with a type-level override, using a type that is
# expected to be used in multiple places
class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }

其次，如果使用 Annotated[]，声明式将从左侧类型中提取完整的 mapped_column() 定义，方法是将 mapped_column() 构造作为任何参数传递给 Annotated[] 构造（感谢 @adriangb01 阐述了这个想法）。此功能可能会在未来的版本中扩展，以包括 relationship()、composite() 和其他构造，但目前仅限于 mapped_column()。下面的示例除了我们的 str50 示例之外，还添加了额外的 Annotated 类型，以说明此功能。

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

# declarative base from previous example
str50 = Annotated[str, 50]


class Base(DeclarativeBase):
    type_annotation_map = {
        str50: String(50),
    }


# set up mapped_column() overrides, using whole column styles that are
# expected to be used in multiple places
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk]
    name: Mapped[str50]
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk]
    email_address: Mapped[str50]
    user_id: Mapped[user_fk]
    user: Mapped["User"] = relationship(back_populates="addresses")

在上面，使用 Mapped[str50]、Mapped[intpk] 或 Mapped[user_fk] 映射的列从 registry.type_annotation_map 以及 Annotated 构造中直接提取信息，以便重用预先建立的类型和列配置。

可选步骤 - 将映射类转换为 dataclasses

我们可以将映射类转换为 dataclasses，其中一个关键优势是我们可以构建一个严格类型化的 __init__() 方法，该方法具有显式的位置参数、仅关键字参数和默认参数，更不用说我们免费获得了诸如 __str__() 和 __repr__() 之类的方法。下一节 对映射为 ORM 模型的 Dataclasses 的原生支持 进一步说明了上述模型的转换。

从第 3 步开始支持类型注解

通过上述示例，从“第 3 步”开始的任何示例都将包括模型的属性已类型化，并将填充到 select()、Query 和 Row 对象。

# (variable) stmt: Select[Tuple[int, str]]
stmt = select(User.id, User.name)

with Session(e) as sess:
    for row in sess.execute(stmt):
        # (variable) row: Row[Tuple[int, str]]
        print(row)

    # (variable) users: Sequence[User]
    users = sess.scalars(select(User)).all()

    # (variable) users_legacy: List[User]
    users_legacy = sess.query(User).all()

另请参阅

带有 mapped_column() 的声明式表 - 更新了声明式文档，用于 Table 列的声明式生成和映射。

使用旧版 Mypy 类型化模型

使用 Mypy 插件 和显式注解的 SQLAlchemy 应用程序，如果注解中未使用 Mapped，则在新系统下会遇到错误，因为当使用诸如 relationship() 之类的构造时，此类注解会被标记为错误。

迁移到 2.0 第六步 - 向显式类型化的 ORM 模型添加 __allow_unmapped__ 章节说明了如何暂时禁用为使用显式注解的旧版 ORM 模型引发这些错误。

另请参阅

迁移到 2.0 第六步 - 向显式类型化的 ORM 模型添加 __allow_unmapped__

对映射为 ORM 模型的 Dataclasses 的原生支持

上面在 ORM 声明式模型 中介绍的新的 ORM 声明式功能引入了新的 mapped_column() 构造，并说明了以类型为中心的映射以及可选的 PEP 593 Annotated 的使用。我们可以通过将其与 Python dataclasses 集成，使映射更进一步。这个新功能是通过 PEP 681 实现的，它允许类型检查器识别与 dataclass 兼容或完全是 dataclass，但通过替代 API 声明的类。

使用 dataclasses 功能，映射类获得了一个 __init__() 方法，该方法支持位置参数以及可选关键字参数的可自定义默认值。如前所述，dataclasses 还生成许多有用的方法，例如 __str__()、__eq__()。Dataclass 序列化方法（例如 dataclasses.asdict() 和 dataclasses.astuple()）也有效，但目前不适用于自引用结构，这使得它们不太适合具有双向关系的映射。

SQLAlchemy 当前的集成方法将用户定义的类转换为真正的 dataclass 以提供运行时功能；该功能利用了 SQLAlchemy 1.4 中在 Python Dataclasses，attrs 支持声明式、命令式映射 中引入的现有 dataclass 功能，以生成具有完全集成配置风格的等效运行时映射，并且类型化也比以前的方法更正确。

为了支持符合 PEP 681 的 dataclasses，ORM 构造（如 mapped_column() 和 relationship()）接受额外的 PEP 681 参数 init、default 和 default_factory，这些参数将传递给 dataclass 创建过程。这些参数目前必须在右侧的显式指令中存在，就像它们将与 dataclasses.field() 一起使用一样；它们目前不能是左侧 Annotated 构造的局部参数。为了支持方便地使用 Annotated，同时仍然支持 dataclass 配置，mapped_column() 可以将最少的右侧参数集与位于 Annotated 构造左侧的现有 mapped_column() 构造的参数合并，以便在下面看到的那样，保持大部分简洁性。

为了使用类继承启用 dataclasses，我们使用 MappedAsDataclass 混入，可以直接在每个类上，也可以在 Base 类上使用，如下所示，我们在其中进一步修改了 ORM 声明式模型 的“第 5 步”中的示例映射。

from typing_extensions import Annotated
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(MappedAsDataclass, DeclarativeBase):
    """subclasses will be converted to dataclasses"""


intpk = Annotated[int, mapped_column(primary_key=True)]
str30 = Annotated[str, mapped_column(String(30))]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[intpk] = mapped_column(init=False)
    name: Mapped[str30]
    fullname: Mapped[Optional[str]] = mapped_column(default=None)
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", default_factory=list
    )


class Address(Base):
    __tablename__ = "address"

    id: Mapped[intpk] = mapped_column(init=False)
    email_address: Mapped[str]
    user_id: Mapped[user_fk] = mapped_column(init=False)
    user: Mapped["User"] = relationship(back_populates="addresses", default=None)

上面的映射已直接在每个映射类上使用了 @dataclasses.dataclass 装饰器，同时设置了声明式映射，在内部设置了每个 dataclasses.field() 指令，如指示的那样。User / Address 结构可以使用配置的位置参数来创建。

>>> u1 = User("username", fullname="full name", addresses=[Address("email@address")])
>>> u1
User(id=None, name='username', fullname='full name', addresses=[Address(id=None, email_address='email@address', user_id=None, user=...)])

另请参阅

声明式 Dataclass 映射

针对除 MySQL 之外的所有后端，现在实现了优化的 ORM 批量插入

在 1.4 系列中引入并在 ORM 批量插入，psycopg2 现在在大多数情况下使用 RETURNING 批量处理语句 中描述的显着性能提升现已推广到所有支持 RETURNING 的包含的后端，即除 MySQL 之外的所有后端：SQLite、MariaDB、PostgreSQL（所有驱动程序）和 Oracle；SQL Server 支持，但在 2.0.9 版本中暂时禁用 [1]。虽然最初的功能对于 psycopg2 驱动程序最为关键，否则在使用 cursor.executemany() 时会遇到严重的性能问题，但此更改对于其他 PostgreSQL 驱动程序（如 asyncpg）也很关键，因为当使用 RETURNING 时，单语句 INSERT 语句仍然慢得无法接受，并且当使用 SQL Server 时，无论是否使用 RETURNING，INSERT 语句的 executemany 速度似乎也很慢。

新功能的性能为基本上每个驱动程序在插入没有预先分配主键值的 ORM 对象时提供了几乎全面的数量级性能提升，如下表所示，在大多数情况下，具体取决于 RETURNING 的使用，而 RETURNING 通常在 executemany() 中不受支持。

psycopg2 “快速执行助手”方法包括将带有单个参数集的 INSERT..RETURNING 语句转换为插入许多参数集的单个语句，使用多个 “VALUES…” 子句，以便它可以一次容纳许多参数集。然后，参数集通常分批分组为 1000 个或类似的组，以便没有单个 INSERT 语句过大，然后为每批参数调用 INSERT 语句，而不是为每个单独的参数集调用。主键值和服务器默认值由 RETURNING 返回，这继续有效，因为每个语句执行都是使用 cursor.execute() 而不是 cursor.executemany() 调用的。

这允许在一个语句中插入多行，同时还能够返回新生成的主键值以及 SQL 和服务器默认值。SQLAlchemy 历史上一直需要为每个参数集调用一个语句，因为它依赖于 Python DBAPI 功能，例如不支持多行的 cursor.lastrowid。

由于大多数数据库现在都提供 RETURNING（MySQL 是明显的例外，因为 MariaDB 支持它），因此新的更改将 psycopg2 “快速执行助手”方法推广到所有支持 RETURNING 的方言，现在包括 SQlite 和 MariaDB，对于这些方言，没有其他 “executemany plus RETURNING” 的方法是可能的，其中包括 SQLite、MariaDB 和所有 PG 驱动程序。用于 Oracle 支持的 cx_Oracle 和 oracledb 驱动程序原生支持带有 executemany 的 RETURNING，这也已实现以提供等效的性能改进。随着 SQLite 和 MariaDB 现在提供 RETURNING 支持，ORM 使用 cursor.lastrowid 几乎成为过去，只有 MySQL 仍然依赖它。

对于不使用 RETURNING 的 INSERT 语句，大多数后端使用传统的 executemany() 行为，但当前的例外是 psycopg2，它总体上具有非常慢的 executemany() 性能，并且仍然可以通过 “insertmanyvalues” 方法得到改进。

基准测试

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

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

此功能改进的操作包括：

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

@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

下表总结了 SQLAlchemy 最新 1.4 系列与 2.0 相比的性能测量结果，两者都运行相同的测试。

驱动程序

SQLA 1.4 时间（秒）

SQLA 2.0 时间（秒）

sqlite+pysqlite2（内存）

6.204843

3.554856

postgresql+asyncpg（网络）

88.292285

4.561492

postgresql+psycopg（网络）

N/A (psycopg3)

4.861368

mssql+pyodbc（网络）

158.396667

4.825139

oracle+cx_Oracle（网络）

92.603953

4.809520

mariadb+mysqldb（网络）

71.705197

4.075377

注意

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

驱动程序

SQLA 1.4 时间（秒）

SQLA 2.0 时间（秒）

postgresql+psycopg2（网络）

4.704876

4.699883

mysql+mysqldb（网络）

77.281997

76.132995

更改摘要

以下要点列出了 2.0 中为使所有驱动程序达到此状态而进行的个别更改：

  • 为 SQLite 实现了 RETURNING - #6195

  • 为 MariaDB 实现了 RETURNING - #7011

  • 修复了 Oracle 的多行 RETURNING - #6245

  • 使 insert() executemany() 支持尽可能多的方言的 RETURNING，通常使用 VALUES() - #6047

  • 当为不支持的后端使用带有 executemany 的 RETURNING 时发出警告（目前没有 RETURNING 后端有此限制）- #7907

  • ORM Mapper.eager_defaults 参数现在默认为新的设置 "auto"，当使用的后端支持带有 “insertmanyvalues” 的 RETURNING 时，它将自动为 INSERT 语句启用 “eager defaults”。有关文档，请参阅 获取服务器生成的默认值。

另请参阅

INSERT 语句的 “Insert Many Values” 行为 - 有关新功能的文档和背景信息，以及如何配置它。

启用 ORM 的 Insert、Upsert、Update 和 Delete 语句，带有 ORM RETURNING

SQLAlchemy 1.4 将旧版 Query 对象的功能移植到 2.0 风格 执行，这意味着 Select 构造可以传递给 Session.execute() 以传递 ORM 结果。还添加了对 Update 和 Delete 的支持，以便传递给 Session.execute()，在某种程度上，它们可以提供 Query.update() 和 Query.delete() 的实现。

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

批量插入并返回 (RETURNING)

Insert 可以传递给 Session.execute()，无论是否带有 Insert.returning()，当与单独的参数列表一起传递时，将调用与先前由 Session.bulk_insert_mappings() 实现的相同过程，并具有额外的增强功能。这将优化行的批处理，利用新的 fast insertmany 功能，同时还增加了对异构参数集和多表映射（如连接表继承）的支持

>>> 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 表达式。此外，还支持 upsert 变体，例如 SQLite、PostgreSQL 和 MariaDB 的变体。这些语句现在可以包含带有列表达式或完整 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 批量插入 (INSERT) 和每行 SQL 表达式

ORM “upsert” 语句

ORM UPDATE / DELETE 和 WHERE … RETURNING

SQLAlchemy 1.4 还对与 update() 和 delete() 构造一起使用时 RETURNING 功能提供了一些有限的支持，当与 Session.execute() 一起使用时。此支持现在已升级为完全原生支持，包括无论是否显式使用 RETURNING，fetch 同步策略也可以继续进行

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

另请参阅

ORM UPDATE 和 DELETE 以及自定义 WHERE 条件

将 RETURNING 与 UPDATE/DELETE 和自定义 WHERE 条件一起使用

改进的 synchronize_session 对 ORM UPDATE / DELETE 的行为

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

另请参阅

选择同步策略

变更摘要

列出的关于带有 RETURNING 功能的新 ORM DML 的 issue

  • 转换 ORM 级别的 insert() 以解释 ORM 上下文中的 values() - #7864

  • 评估 dml.returning(Entity) 交付 ORM 表达式的可行性，自动应用 select().from_statement 等价物 - #7865

  • 给定 ORM insert，尝试沿用批量方法，关于继承 - #8360

新的“仅写入”关系策略取代了“动态”策略

lazy="dynamic" 加载器策略变为旧版，因为它被硬编码为使用旧版的 Query。此加载器策略既不与 asyncio 兼容，并且还具有许多隐式迭代其内容的行为，这违背了 “dynamic” 关系的最初目的，即用于不应在任何时候隐式完全加载到内存中的非常大的集合。

“dynamic” 策略现在被新的策略 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"))
)

更大的区别在于数据库加载方面，集合无法直接从数据库加载对象；相反，SQL 构造方法（例如 WriteOnlyCollection.select()）用于生成 SQL 构造（例如 Select），然后使用 2.0 风格 执行这些构造，以显式方式加载所需的对象

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

WriteOnlyCollection 还与新的 ORM 批量 dml 功能集成，包括对批量 INSERT 和带有 WHERE 条件的 UPDATE/DELETE 的支持，所有这些都包括 RETURNING 支持。请参阅 只写关系 中的完整文档。

另请参阅

只写关系

新的 pep-484 / 类型注解映射支持动态关系

即使 “dynamic” 关系在 2.0 中已成为旧版，但由于这些模式预计会有很长的生命周期，因此现在为 “dynamic” 关系添加了 类型注解映射 支持，其方式与新的 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())

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

另请参阅

动态关系加载器

#7123

安装现在完全启用 pep-517

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

#7311

C 扩展现在已移植到 Cython

SQLAlchemy C 扩展已被用 Cython 编写的全新扩展所取代。虽然 Cython 在 2010 年首次创建 C 扩展时就已进行评估，但今天使用的 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 发行版中预构建，这些发行版可从 PyPi 自动提供给 pip。手动构建说明也未更改，除了 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 上运行，在这里我们使用它来说明调用 metadata.reflect() 以通过本地网络连接一次反射 250 个 Table 对象的性能差异

方言

操作

SQLA 1.4 时间（秒）

SQLA 2.0 时间（秒）

postgresql+psycopg2

metadata.reflect()，250 个表

8.2

3.3

oracle+cx_oracle

metadata.reflect()，250 个表

60.4

6.8

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() 方法是缓存的，从而导致两种方法类型之间的结果不一致。

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

Inspector() 的新方法和改进

#4379

对 psycopg 3（又名 “psycopg”）的方言支持

增加了对 psycopg 3 DBAPI 的方言支持，尽管数字为 “3”，但现在的包名称为 psycopg，取代了之前的 psycopg2 包，后者暂时仍是 postgresql 方言的 “默认” 驱动程序。psycopg 是 PostgreSQL 的完全重新设计和现代化的数据库适配器，它支持诸如预处理语句以及 Python asyncio 等概念。

psycopg 是 SQLAlchemy 支持的第一个 DBAPI，它同时提供 pep-249 同步 API 和 asyncio 驱动程序。相同的 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 之类的构造基于 DDLElement.execute_if() 方法接受的相同类型的条件，针对给定的 Table 有条件地呈现。在下面的示例中，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 的一部分）。

当将 literal_binds 与 PostgreSQL、MySQL、MariaDB、MSSQL、Oracle 方言提供的 SQL 编译器一起使用时，默认渲染是修改后的 ISO-8601 渲染（即，ISO-8601，其中 T 转换为空格）。对于 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 中发生的行为变更，这些变更不属于主要的 1.4->2.0 迁移路径；此处的变更预计不会对向后兼容性产生重大影响。

Session 的新事务加入模式

“将外部事务加入到 Session 中” 的行为已得到修订和改进，允许显式控制 Session 将如何适应传入的 Connection，该连接已具有事务，并且可能已建立保存点。新的参数 Session.join_transaction_mode 包括一系列选项值，这些选项值可以通过多种方式适应现有事务，最重要的是允许 Session 以完全事务性风格使用专门的保存点进行操作，同时在所有情况下都保持外部发起的事务未提交且处于活动状态，从而允许测试套件回滚测试中发生的所有更改。

这允许的主要改进是 将 Session 加入到外部事务中（例如用于测试套件） 中记录的配方（该配方也从 SQLAlchemy 1.3 更改为 1.4）现在已得到简化，不再需要显式使用事件处理程序或任何显式保存点的提及；通过使用 join_transaction_mode="create_savepoint"，Session 将永远不会影响传入事务的状态，而是将创建一个保存点（即 “嵌套事务”）作为其根事务。

以下说明了 将 Session 加入到外部事务中（例如用于测试套件） 中给出的示例的一部分；有关完整示例，请参见该部分

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection, selecting
        # "create_savepoint" join_transaction_mode
        self.session = Session(
            bind=self.connection, join_transaction_mode="create_savepoint"
        )

    def tearDown(self):
        self.session.close()

        # rollback non-ORM transaction
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

为 Session.join_transaction_mode 选择的默认模式是 "conditional_savepoint"，如果给定的 Connection 本身已处于保存点 (savepoint) 上，则使用 "create_savepoint" 行为。如果给定的 Connection 处于事务中但不在保存点上，则 Session 将传播 “rollback” 调用，但不传播 “commit” 调用，但不会自行启动新的保存点。默认选择此行为是为了最大限度地兼容旧版本的 SQLAlchemy，并且除非给定的驱动程序已在使用保存点，否则它不会启动新的 SAVEPOINT，因为对 SAVEPOINT 的支持不仅因特定的后端和驱动程序而异，而且还因配置而异。

以下示例说明了在 SQLAlchemy 1.3 中有效，在 SQLAlchemy 1.4 中失效，现在在 SQLAlchemy 2.0 中恢复的情况

engine = create_engine("...")

# setup outer connection with a transaction and a SAVEPOINT
conn = engine.connect()
trans = conn.begin()
nested = conn.begin_nested()

# bind a Session to that connection and operate upon it, including
# a commit
session = Session(conn)
session.connection()
session.commit()
session.close()

# assert both SAVEPOINT and transaction remain active
assert nested.is_active
nested.rollback()
trans.rollback()

在上面，一个 Session 被加入到一个 Connection，该连接已在其上启动了一个保存点；在 Session 与事务交互后，这两个单元的状态保持不变。在 SQLAlchemy 1.3 中，上述情况有效，因为 Session 将在 Connection 上启动一个“子事务”，这将允许外部保存点/事务在上述简单情况下保持不受影响。由于子事务在 1.4 中已被弃用，并且现在在 2.0 中已被删除，因此这种行为不再可用。新的默认行为通过使用真正的、第二个 SAVEPOINT 改进了“子事务”的行为，因此即使调用 Session.rollback() 也可以防止 Session “突破”到外部启动的 SAVEPOINT 或事务中。

但是，将已启动事务的 Connection 加入到 Session 的新代码应显式选择 Session.join_transaction_mode，以便显式定义所需的行为。

#9015

str(engine.url) 默认会混淆密码

为了避免数据库密码泄露，现在默认情况下，在 URL 上调用 str() 将启用密码混淆功能。以前，此混淆会应用于 __repr__() 调用，但不适用于 __str__()。此更改将影响尝试调用 create_engine() 的应用程序和测试套件，这些应用程序和测试套件使用来自另一个引擎的字符串化 URL，例如：

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(str(e1.url))

上面的引擎 e2 将不具有正确的密码；它将具有混淆后的字符串 "***"。

上述模式的首选方法是直接传递 URL 对象，无需字符串化

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> e2 = create_engine(e1.url)

否则，对于带有明文密码的字符串化 URL，请使用 URL.render_as_string() 方法，并将 URL.render_as_string.hide_password 参数设置为 False

>>> e1 = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> url_string = e1.url.render_as_string(hide_password=False)
>>> e2 = create_engine(url_string)

#8567

对具有相同名称、键的 Table 对象中 Column 的替换实施更严格的规则

对于将 Column 对象追加到 Table 对象，现在实施了更严格的规则，既将以前的一些弃用警告升级为异常，又防止了以前某些会导致表中出现重复列的情况，当 Table.extend_existing 设置为 True 时，无论是在程序化的 Table 构建期间还是在反射操作期间。

  • 在任何情况下，Table 对象都不应有两个或多个具有相同名称的 Column 对象，无论它们的 .key 是什么。已识别并修复了一个仍然可能发生这种情况的边缘情况。

  • 将 Column 添加到 Table，如果该表已存在具有相同名称或键的 Column，则总是会引发 DuplicateColumnError（2.0.0b4 中的 ArgumentError 的新子类），除非存在其他参数；对于 Table.append_column()，使用 Table.append_column.replace_existing，对于构造与现有表同名的 Table，无论是否使用反射，都使用 Table.extend_existing。以前，对于这种情况，会发出弃用警告。

  • 现在，如果创建的 Table 包含 Table.extend_existing，并且传入的 Column 没有单独的 Column.key，则会发出警告，因为它会完全替换具有键的现有 Column，这表明该操作并非用户所期望的。这种情况尤其可能在二级反射步骤中发生，例如 metadata.reflect(extend_existing=True)。警告建议将 Table.autoload_replace 参数设置为 False 以防止这种情况发生。以前，在 1.4 及更早版本中，传入的列将**额外**添加到现有列中。这是一个错误，并且是 2.0 中的行为更改（自 2.0.0b4 起），因为发生这种情况时，之前的键将**不再存在**于列集合中。

#8925

ORM 声明式以不同的方式应用列顺序；使用 sort_order 控制行为

声明式更改了对来自 mixin 或抽象基类的映射列以及声明类本身的列进行排序的系统，以便将声明类中的列放在前面，然后是 mixin 列。以下映射：

class Foo:
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)


class Bar:
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)


class Model(Base, Foo, Bar):
    id = mapped_column(Integer, primary_key=True)
    __tablename__ = "model"

在 1.4 上生成如下的 CREATE TABLE：

CREATE TABLE model (
  col1 INTEGER,
  col3 INTEGER,
  col2 INTEGER,
  col4 INTEGER,
  id INTEGER NOT NULL,
  PRIMARY KEY (id)
)

而在 2.0 上生成：

CREATE TABLE model (
  id INTEGER NOT NULL,
  col1 INTEGER,
  col3 INTEGER,
  col2 INTEGER,
  col4 INTEGER,
  PRIMARY KEY (id)
)

对于上面的特定情况，这可以被视为一种改进，因为 Model 上的主键列现在位于通常首选的位置。但是，对于以另一种方式定义模型的应用程序来说，这并不能带来安慰，因为：

class Foo:
    id = mapped_column(Integer, primary_key=True)
    col1 = mapped_column(Integer)
    col3 = mapped_column(Integer)


class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

现在生成如下的 CREATE TABLE 输出：

CREATE TABLE model (
  col2 INTEGER,
  col4 INTEGER,
  id INTEGER NOT NULL,
  col1 INTEGER,
  col3 INTEGER,
  PRIMARY KEY (id)
)

为了解决这个问题，SQLAlchemy 2.0.4 在 mapped_column() 上引入了一个名为 mapped_column.sort_order 的新参数，它是一个整数值，默认为 0，可以设置为正值或负值，以便将列放置在其他列之前或之后，如下例所示：

class Foo:
    id = mapped_column(Integer, primary_key=True, sort_order=-10)
    col1 = mapped_column(Integer, sort_order=-1)
    col3 = mapped_column(Integer)


class Model(Foo, Base):
    col2 = mapped_column(Integer)
    col4 = mapped_column(Integer)
    __tablename__ = "model"

上面的模型将 “id” 放在所有其他列之前，将 “col1” 放在 “id” 之后

CREATE TABLE model (
  id INTEGER NOT NULL,
  col1 INTEGER,
  col2 INTEGER,
  col4 INTEGER,
  col3 INTEGER,
  PRIMARY KEY (id)
)

未来的 SQLAlchemy 版本可能会选择为 mapped_column 构造提供显式的排序提示，因为此排序是 ORM 特有的。

Sequence 构造恢复为没有任何显式默认 “start” 值；影响 MS SQL Server

在 SQLAlchemy 1.4 之前，如果未指定其他参数，Sequence 构造只会发出简单的 CREATE SEQUENCE DDL：

>>> # SQLAlchemy 1.3 (and 2.0)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))

CREATE SEQUENCE my_seq

但是，由于为 MS SQL Server 添加了 Sequence 支持，其中默认起始值不方便地设置为 -2**63，因此 1.4 版本决定如果未另行提供 Sequence.start，则默认 DDL 发出起始值为 1：

>>> # SQLAlchemy 1.4 (only)
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq")))

CREATE SEQUENCE my_seq START WITH 1

此更改引入了其他复杂性，包括当包含 Sequence.min_value 参数时，此默认值 1 实际上应默认为 Sequence.min_value 声明的值，否则低于 start_value 的 min_value 可能会被视为矛盾。由于对这个问题的研究开始变得有点像其他各种边缘情况的兔子洞，我们决定撤消此更改，并恢复 Sequence 的原始行为，即不发表意见，只发出 CREATE SEQUENCE，允许数据库本身决定 SEQUENCE 的各种参数应如何相互作用。

因此，为了确保所有后端上的起始值都为 1，**可以显式指示起始值为 1**，如下所示：

>>> # All SQLAlchemy versions
>>> from sqlalchemy import Sequence
>>> from sqlalchemy.schema import CreateSequence
>>> print(CreateSequence(Sequence("my_seq", start=1)))

CREATE SEQUENCE my_seq START WITH 1

除此之外，对于包括 PostgreSQL、Oracle、SQL Server 在内的现代后端上的整数主键的自动生成，应首选 Identity 构造，它在 1.4 和 2.0 中的工作方式相同，行为没有变化。

#7211

“with_variant()” 克隆原始 TypeEngine，而不是更改类型

TypeEngine.with_variant() 方法用于将备用数据库特定行为应用于特定类型，现在返回原始 TypeEngine 对象的副本，并将变体信息存储在内部，而不是将其包装在 Variant 类中。

虽然之前的 Variant 方法能够使用动态属性 getter 维护原始类型的所有 Python 行为，但此处的改进在于，当调用变体时，返回的类型仍然是原始类型的实例，这可以更顺畅地与 mypy 和 pylance 等类型检查器一起使用。给定如下程序：

import typing

from sqlalchemy import String
from sqlalchemy.dialects.mysql import VARCHAR

type_ = String(255).with_variant(VARCHAR(255, charset="utf8mb4"), "mysql", "mariadb")

if typing.TYPE_CHECKING:
    reveal_type(type_)

像 pyright 这样的类型检查器现在会将类型报告为：

info: Type of "type_" is "String"

此外，如上所示，可以为单个类型传递多个方言名称，特别是这对于 "mysql" 和 "mariadb" 方言对很有帮助，从 SQLAlchemy 1.4 开始，它们被视为单独的方言。

#6980

Python 除法运算符对所有后端执行真除法；添加了向下取整除法

Core 表达式语言现在支持 “真除法”（即 / Python 运算符）和 “向下取整除法”（即 // Python 运算符），包括后端特定的行为，以规范化这方面不同的数据库。

给定对两个整数值执行 “真除法” 操作：

expr = literal(5, Integer) / literal(10, Integer)

例如，PostgreSQL 上的 SQL 除法运算符在对整数使用时通常充当 “向下取整除法”，这意味着上面的结果将返回整数 “0”。对于此后端和类似的后端，SQLAlchemy 现在使用一种等效于以下形式的 SQL 进行渲染：

%(param_1)s / CAST(%(param_2)s AS NUMERIC)

使用 param_1=5，param_2=10，以便返回表达式的类型为 NUMERIC，通常为 Python 值 decimal.Decimal("0.5")。

给定对两个整数值执行 “向下取整除法” 操作：

expr = literal(5, Integer) // literal(10, Integer)

例如，MySQL 和 Oracle 上的 SQL 除法运算符在对整数使用时通常充当 “真除法”，这意味着上面的结果将返回浮点值 “0.5”。对于这些后端和类似的后端，SQLAlchemy 现在使用一种等效于以下形式的 SQL 进行渲染：

FLOOR(%(param_1)s / %(param_2)s)

使用 param_1=5, param_2=10, 以便返回表达式的类型为 INTEGER，作为 Python 值 0。

如果使用 PostgreSQL、SQL Server 或 SQLite 的应用程序依赖于 Python “truediv” 运算符在所有情况下都返回整数值，则此处会发生向后不兼容的更改。依赖于此行为的应用程序应改为对这些操作使用 Python “向下取整除法” 运算符 //，或者为了在使用以前的 SQLAlchemy 版本时的向前兼容性，使用 floor 函数：

expr = func.floor(literal(5, Integer) / literal(10, Integer))

为了提供后端无关的向下取整除法，在 2.0 之前的任何 SQLAlchemy 版本上都需要上述形式。

#4926

当检测到非法并发或重入访问时，Session 主动引发异常

现在，Session 可以捕获更多与多线程或其他并发场景中的非法并发状态更改以及执行意外状态更改的事件钩子相关的错误。

当在多个线程中同时使用 Session 时，已知会发生一个错误：AttributeError: 'NoneType' object has no attribute 'twophase'，这完全令人费解。当一个线程调用 Session.commit() 时，会发生此错误，该调用在内部调用 SessionTransaction.close() 方法以结束事务上下文，同时另一个线程正在从 Session.execute() 运行查询。在 Session.execute() 中，为当前事务获取数据库连接的内部方法首先断言会话是 “active”，但在此断言通过后，并发调用 Session.close() 会干扰此状态，从而导致上述未定义的条件。

此更改将保护应用于围绕 SessionTransaction 对象的所有状态更改方法，以便在上述情况下，Session.commit() 方法将改为失败，因为它将尝试将状态更改为已在进行中的方法不允许的状态，该方法想要获取当前连接以运行数据库查询。

使用 #7433 中所示的测试脚本，之前的错误情况如下所示：

Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test3.py", line 30, in worker
    sess.execute(select(A)).all()
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1691, in execute
    conn = self._connection_for_bind(bind)
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1532, in _connection_for_bind
    return self._transaction._connection_for_bind(
File "/home/classic/tmp/sqlalchemy/lib/sqlalchemy/orm/session.py", line 754, in _connection_for_bind
    if self.session.twophase and self._parent is None:
AttributeError: 'NoneType' object has no attribute 'twophase'

其中 _connection_for_bind() 方法无法继续，因为并发访问使其处于无效状态。使用新方法，状态更改的发起者会改为抛出错误：

File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1785, in close
   self._close_impl(invalidate=False)
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1827, in _close_impl
   transaction.close(invalidate)
File "<string>", line 2, in close
File "/home/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 506, in _go
   raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Method 'close()' can't be called here;
method '_connection_for_bind()' is already in progress and this would cause
an unexpected state change to symbol('CLOSED')

状态转换检查有意不使用显式锁来检测并发线程活动，而是依赖于简单的属性设置/值测试操作，这些操作在发生意外并发更改时自然会失败。这样做的理由是，该方法可以检测到完全在单个线程中发生的非法状态更改，例如在会话事务事件上运行的事件处理程序调用了意外的状态更改方法，或者在使用 asyncio 时，如果特定的 Session 在多个 asyncio 任务之间共享，以及在使用修补式并发方法（如 gevent）时。

#7433

SQLite 方言对基于文件的数据库使用 QueuePool

当使用基于文件的数据库时，SQLite 方言现在默认为 QueuePool。这与将 check_same_thread 参数设置为 False 一起设置。据观察，以前默认使用 NullPool（在释放后不保留数据库连接）的方法实际上具有可衡量的负面性能影响。与往常一样，可以通过 create_engine.poolclass 参数自定义池类。

在版本 2.0.38 中更改: - aiosqlite 方言也进行了等效的更改，使用 AsyncAdaptedQueuePool 而不是 NullPool。aiosqlite 方言在最初的更改中被错误地排除在外。

另请参阅

线程/连接池行为

#7490

新的 Oracle FLOAT 类型具有二进制精度；不直接接受十进制精度

Oracle 方言中添加了一个新的数据类型 FLOAT，以配合 Double 和数据库特定的 DOUBLE、DOUBLE_PRECISION 和 REAL 数据类型的添加。Oracle 的 FLOAT 接受所谓的 “二进制精度” 参数，根据 Oracle 文档，该参数大致是标准 “精度” 值除以 0.3103。

from sqlalchemy.dialects import oracle

Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

二进制精度值 126 与使用 DOUBLE_PRECISION 数据类型同义，值 63 等效于使用 REAL 数据类型。其他精度值特定于 FLOAT 类型本身。

SQLAlchemy Float 数据类型也接受 “precision” 参数，但这指的是十进制精度，Oracle 不接受。Oracle 方言现在不会尝试猜测转换，而是在对 Oracle 后端使用带有精度值的 Float 时引发信息丰富的错误。要指定具有显式精度值以支持后端的 Float 数据类型，同时支持其他后端，请使用 TypeEngine.with_variant() 方法，如下所示：

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle

Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)

新的 RANGE / MULTIRANGE 支持以及 PostgreSQL 后端的更改

已为 psycopg2、psycopg3 和 asyncpg 方言完全实现了 RANGE / MULTIRANGE 支持。新的支持使用新的 SQLAlchemy 特定的 Range 对象，该对象与不同的后端无关，并且不需要使用后端特定的导入或扩展步骤。对于 multirange 支持，使用 Range 对象的列表。

应修改使用以前的 psycopg2 特定类型的代码以使用 Range，它提供兼容的接口。

Range 对象还具有镜像 PostgreSQL 的比较支持。目前已实现 Range.contains() 和 Range.contained_by() 方法，它们的工作方式与 PostgreSQL @> 和 <@ 相同。未来版本可能会添加额外的运算符支持。

有关使用新功能的背景信息，请参阅 范围和多范围类型 的文档。

另请参阅

范围和多范围类型

#7156 #8706

PostgreSQL 上的 match() 运算符使用 plainto_tsquery() 而不是 to_tsquery()

Operators.match() 函数现在在 PostgreSQL 后端上呈现 col @@ plainto_tsquery(expr)，而不是 col @@ to_tsquery()。plainto_tsquery() 接受纯文本，而 to_tsquery() 接受专门的查询符号，因此与其他后端的交叉兼容性较差。

所有 PostgreSQL 搜索函数和操作符都可以通过使用 func 生成 PostgreSQL 特定的函数和 Operators.bool_op() ( Operators.op() 的布尔类型版本) 生成任意操作符来使用，这与之前的版本中的使用方式相同。请参阅 全文搜索 中的示例。

现有的 SQLAlchemy 项目如果使用了 Operators.match() 中的 PG 特定指令，则应直接使用 func.to_tsquery()。要以与 1.4 版本中完全相同的形式渲染 SQL，请参阅 简单纯文本匹配与 match() 中的版本说明。

#7086