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

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

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

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

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

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

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"

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

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

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  model  (
  col2  INTEGER,
  col4  INTEGER,
  id  INTEGER  NOT  NULL,
  col1  INTEGER,
  col3  INTEGER,
  PRIMARY  KEY  (id)
)

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"

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

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

>>> # 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 

>>> # 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 

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

info: Type of "type_" is "String"

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

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

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

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

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

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'

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

from sqlalchemy.dialects import oracle
Table("some_table", metadata, Column("value", oracle.FLOAT(126)))

from sqlalchemy.types import Float
from sqlalchemy.dialects import oracle
Table(
    "some_table",
    metadata,
    Column("value", Float(5).with_variant(oracle.FLOAT(16), "oracle")),
)
```### PostgreSQL 后端的新 RANGE / MULTIRANGE 支持和更改
RANGE / MULTIRANGE 支持已完全实现为 psycopg2、psycopg3 和 asyncpg 方言。新支持使用了一个新的 SQLAlchemy 特定的 `Range` 对象，它对不同的后端是不可知的，不需要使用后端特定的导入或扩展步骤。对于多范围支持，使用 `Range` 对象的列表。
之前使用的特定于 psycopg2 的类型的代码应修改为使用`Range`，它提供了兼容的接口。
`Range` 对象还具有与 PostgreSQL 相同的比较支持。到目前为止已实现了 `Range.contains()` 和 `Range.contained_by()` 方法，它们的工作方式与 PostgreSQL 的 `@>` 和 `<@` 相同。未来的版本可能会添加其他运算符支持。
请参阅范围和多范围类型处的文档，了解如何使用新功能的背景信息。
另请参阅
范围和多范围类型
[#7156](https://www.sqlalchemy.org/trac/ticket/7156) [#8706](https://www.sqlalchemy.org/trac/ticket/8706)  ### 在 PostgreSQL 上使用`match()`操作符使用`plainto_tsquery()`而不是`to_tsquery()`
`Operators.match()`函数现在在 PostgreSQL 后端上呈现为`col @@ plainto_tsquery(expr)`，而不是`col @@ to_tsquery()`。`plainto_tsquery()`接受纯文本，而`to_tsquery()`接受专门的查询符号，因此与其他后端的兼容性较差。
通过使用`func`生成 PostgreSQL 特定函数和`Operators.bool_op()`（`Operators.op()`的布尔类型版本）来生成任意运算符，以与之前版本相同的方式，所有 PostgreSQL 搜索函数和操作符都可用。请参阅全文搜索中的示例。
现有的 SQLAlchemy 项目如果在`Operators.match()`中使用了 PG 特定的指令，应该直接使用`func.to_tsquery()`。要以与 1.4 版本完全相同的形式呈现 SQL，请参阅使用 match()进行简单纯文本匹配的版本说明。
[#7086](https://www.sqlalchemy.org/trac/ticket/7086)
## Core 和 ORM 中的新类型支持-不再使用存根/扩展
与 1.4 版本中通过[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)包提供的临时方法相比，Core 和 ORM 的类型化方法已完全重写。新方法始于 SQLAlchemy 中最基本的元素，即`Column`，更准确地说是底层所有具有类型的 SQL 表达式的`ColumnElement`。然后，这个表达式级别的类型化扩展到语句构造、语句执行和结果集的领域，最终扩展到 ORM，其中新的声明性形式允许完全类型化的 ORM 模型，从语句到结果集全部集成。
提示
对于 2.0 系列，类型支持应被视为**beta 级别**软件。类型详细信息可能会发生变化，但不计划进行重大不兼容更改。
### SQL 表达式/语句/结果集类型化
本节提供了 SQLAlchemy 的新 SQL 表达类型方法的背景和示例，它从基本的`ColumnElement`构造扩展到 SQL 语句和结果集，以及 ORM 映射的领域。
#### 原因和概述
提示
本节是一个架构讨论。 跳转到 SQL 表达类型 - 示例以查看新的类型化外观。
在[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)中，SQL 表达式被类型化为[泛型](https://peps.python.org/pep-0484/#generics)，然后引用`TypeEngine`对象，如`Integer`，`DateTime`，或`String`作为它们的泛型参数（如`Column[Integer]`）。 这本身就是与最初的 Dropbox [sqlalchemy-stubs](https://github.com/dropbox/sqlalchemy-stubs)包不同的地方，其中`Column`及其基本构造直接以 Python 类型为泛型，例如`int`，`datetime`和`str`。 希望由于`Integer` / `DateTime` / `String`本身是对`int` / `datetime` / `str`的泛型，因此有方法可以保持两个级别的信息，并且可以通过`TypeEngine`从列表达式中提取 Python 类型作为中间构造。 但是，事实并非如此，因为[**PEP 484**](https://peps.python.org/pep-0484/)没有足够丰富的功能集使其可行，缺乏诸如[higher kinded TypeVars](https://github.com/python/typing/issues/548)之类的功能。
经过对[当前能力的深入评估](https://github.com/python/typing/discussions/999)，SQLAlchemy 2.0 实现了[**PEP 484**](https://peps.python.org/pep-0484/)的原始智慧，直接将列表达式与 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**](https://peps.python.org/pep-0646/)是第一个试图处理类似元组对象的 pep，[故意在功能上受到限制](https://mail.python.org/archives/list/typing-sig@python.org/message/G2PNHRR32JMFD3JR7ACA2NDKWTDSEPUG/)，并且本身尚不适用于任意元组操作），但已经设计出了一个相当不错的方法，允许基本的`select()` -> `Result` -> `Row`类型化功能，包括 ORM 类，其中在将`Row`对象解包为单独列条目时，添加了一个小的面向类型的访问器，允许各个 Python 值保持与其来源的 SQL 表达式相关联的 Python 类型（翻译：它有效）。
#### SQL 表达式类型化 - 示例
简要介绍了类型行为。注释指示在[vscode](https://code.visualstudio.com/)中悬停在代码上会看到什么（或者使用[reveal_type()](https://mypy.readthedocs.io/en/latest/common_issues.html?highlight=reveal_type#reveal-type)辅助工具时大致会显示什么）：
+   简单的 Python 类型分配给 SQL 表达式
    ```py
    # (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.returning()`的`Insert`，都打包成一个保留每个元素 Python 类型的`Tuple[]`类型。
    ```py
    # (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`对象）。
    ```py
    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()`等方法是正确的。
    ```py
    # (variable) data: Sequence[str]
    data = connection.execute(select(str_col)).scalars().all()
    ```
+   对于 ORM 映射类，上述对返回行构造的支持与其最配套，因为映射类可以列出其成员的特定类型。下面的示例设置了一个使用新的类型感知语法的类，将在下一节中描述：
    ```py
    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"))
    ```
    通过上述映射，属性从语句到结果集一路类型化并表达自己：
    ```py
    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 查询：
    ```py
    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`这样的构造也能正常工作，保持原始映射类的列级属性以及语句期望的返回类型：
    ```py
    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)
    ```
+   核心表（Core Table）目前还没有一个合适的方法来在通过`Table.c`访问时维护`Column`对象的类型。
    由于`Table`被设置为类的实例，并且`Table.c`访问器通常通过名称动态访问`Column`对象，因此尚未为此建立类型化方法；需要一些替代语法。
+   ORM 类、标量等工作得很好。
    选择 ORM 类作为标量或元组的典型用例都有效，无论是 2.0 还是 1.x 风格的查询，都可以获得确切的类型，无论是单独还是包含在适当的容器中，如`Sequence[]`、`List[]`或`Iterator[]`：
    ```py
    # (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](https://github.com/dropbox/sqlalchemy-stubs)或[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)提供的范围，其中标量对象以及元组类型的`Query`对象将保留大多数情况下的结果级别类型：
    ```py
    # (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()
    ```
#### 注意 - 所有存根必须被卸载
类型支持的一个关键警告是**必须卸载所有 SQLAlchemy 存根包**才能使类型化工作。在针对 Python 虚拟环境运行[mypy](https://mypy.readthedocs.io/en/stable/)时，只需卸载这些包。但是，SQLAlchemy 存根包目前也是[typeshed](https://github.com/python/typeshed)的一部分，它本身捆绑在一些类型工具中，如[Pylance](https://github.com/microsoft/pylance-release)，因此在某些情况下可能需要定位这些包的文件并删除它们，如果它们实际上干扰了新的类型化正确工作。
一旦 SQLAlchemy 2.0 以最终状态发布，typeshed 将从其自己的存根源中删除 SQLAlchemy。
### ORM 声明模型
SQLAlchemy 1.4 引入了第一个使用[sqlalchemy2-stubs](https://github.com/sqlalchemy/sqlalchemy2-stubs)和 Mypy Plugin 组合的 SQLAlchemy 本机 ORM 类型支持。在 SQLAlchemy 2.0 中，Mypy 插件**仍然可用，并已更新以与 SQLAlchemy 2.0 的类型系统一起使用**。但是，现在应该将其视为**已弃用**，因为应用程序现在有一条直接的路径来采用新的类型支持，而不使用插件或存根。
#### 概述
新系统的基本方法是，当使用完全声明式模型（即不使用混合声明式或命令式配置，这些配置不变）时，映射列声明首先在运行时通过检查每个属性声明左侧的类型注释来推导，如果存在的话。左手类型注释应该包含在`Mapped`泛型类型中，否则该属性不被视为映射属性。然后属性声明可以引用右侧的`mapped_column()`构造，用于提供有关要生成和映射的`Column`的附加核心级模式信息。如果左侧存在`Mapped`注释，则此右侧声明是可选的；如果左侧没有注释，则`mapped_column()`可以用作`Column`指令的精确替代，其中它将提供更准确（但不精确）的属性类型行为，即使没有注释存在。
这种方法受到 Python [dataclasses](https://docs.python.org/3/library/dataclasses.html)方法的启发，它从左边开始注释，然后允许右边的可选`dataclasses.field()`规范；与 dataclasses 方法的关键区别在于 SQLAlchemy 的方法是严格的**选择加入**，其中使用`Column`的现有映射如果没有任何类型注释，将继续像以往一样工作，并且`mapped_column()`构造可以直接替换`Column`而不需要任何显式类型注释。只有在确切的属性级 Python 类型存在时，才需要使用带有`Mapped`的显式注释。这些注释可以根据需要，按属性基础使用，对于那些特定类型有帮助的属性；使用`mapped_column()`的未注释属性将在实例级别被标记为`Any`。
#### 迁移现有映射
迁移到新的 ORM 方法开始时更加冗长，但随着可用的新功能的充分利用，变得比以前更简洁。以下步骤详细说明了典型的过渡，然后继续说明了一些更多的选项。
##### 第一步 - `declarative_base()`被`DeclarativeBase`取代。
Python 类型中观察到的一个限制是似乎没有能力从函数动态生成一个类，然后被类型工具理解为新类的基础。为了解决这个问题而不使用插件，通常对`declarative_base()`的调用可以替换为使用`DeclarativeBase`类，它产生与通常相同的`Base`对象，只是类型工具理解它：
```py
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass