SqlAlchemy 2.0 中文文档(五十七)(4)

简介: SqlAlchemy 2.0 中文文档(五十七)

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


第二步 - 用mapped_column()替换Column的声明性使用

mapped_column()是一个 ORM-typing 感知构造,可以直接用于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")

我们用mapped_column()替换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];这是因为我们可以将任何列声明为可选的或非可选的,并且没有办法在我们明确地对其进行类型化时不引起类型错误。

但是,在这一步中,我们上述的映射已经为所有属性设置了适当的描述符,并且可以用于查询以及实例级别的操作,所有这些操作都将在mypy –strict mode下通过,而无需插件。

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

这可以应用于所有需要确切类型的属性;可以跳过那些可以保留为Any的属性。为了上下文,我们还说明了在一个relationship()中使用Mapped的情况,我们在这个中间步骤中应用了一个确切的类型。映射在这个中间步骤中将更加冗长,但是通过熟练掌握,这一步可以与后续步骤结合起来更直接地更新映射:

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()QueryResult构造。现在我们可以继续减少映射声明中的冗余部分。

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

所有nullable参数都可以使用Optional[]隐含;在没有Optional[]的情况下,nullable默认为False。所有没有参数的 SQL 类型,如IntegerString,可以仅表示为 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的不同变体,如下所示,使用Annotated str称为str50将表示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()QueryRow对象:

# (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实现的,该特性允许类型检查器识别与数据类兼容或完全是数据类但是通过替代 API 声明的类。

使用数据类特性,映射类获得一个__init__()方法,支持位置参数以及可选关键字参数的可定制默认值。正如前面提到的,数据类还生成许多有用的方法,如__str__()__eq__()。数据类的序列化方法,如dataclasses.asdict()dataclasses.astuple(),也可以使用,但目前不支持自引用结构,这使得它们对具有双向关系的映射不太适用。

SQLAlchemy 的当前集成方法将用户定义的类转换为真正的数据类,以提供运行时功能;该功能利用了  SQLAlchemy 1.4  中引入的现有数据类功能,以在完全集成的配置样式下生成等效的运行时映射,这种映射比以前的方法更正确地进行了类型标记,并且还使用了在 Python  数据类,attrs 支持的情况下,声明式,命令式映射中介绍的特性。

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

为了启用使用类继承的数据类,我们利用了MappedAsDataclass mixin,可以直接在每个类上使用,也可以在Base类上使用,如下面的示例所示,我们进一步修改了“步骤 5”的示例映射来自 ORM 声明模型:

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=...)])

另请参阅

声明性数据类映射

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

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

原理与概述

提示

本节是一个架构讨论。直接跳转到 SQL 表达式类型化 - 示例来查看新的类型化是什么样子的。

sqlalchemy2-stubs 中,SQL 表达式被类型化为泛型,然后引用了一个 TypeEngine 对象,如 IntegerDateTimeString 作为它们的泛型参数(如 Column[Integer])。这本身就是与原始 Dropbox sqlalchemy-stubs 软件包所做的不同,其中 Column 及其基础构造直接是 Python 类型的泛型,比如 intdatetimestr。希望通过 Integer / DateTime / String 本身对 int / datetime / str 的泛型,可以维护两个信息级别,并且可以通过 TypeEngine 作为中间构造从列表达式中提取 Python 类型。然而,情况并非如此,因为PEP 484 没有足够丰富的功能集来实现这一点,缺乏诸如高阶类型变量之类的功能。

经过对当前PEP 484的能力的深度评估,SQLAlchemy 2.0 在这一领域实现了 sqlalchemy-stubs 的原始智慧,并直接将列表达式与 Python 类型进行了链接。这意味着,如果有 SQL 表达式到不同子类型的情况,比如 Column(VARCHAR) vs. Column(Unicode),那么这两种 String 子类型的具体细节并不会随着类型一起传递,但在实践中,这通常不是问题,而且通常情况下,Python 类型是立即存在的,因为它代表了直接存储和接收该列的 Python 数据。

具体来说,这意味着像 Column('id', Integer) 这样的表达式被类型化为 Column[int]。 这允许建立一个可行的 SQLAlchemy 构造 -> Python 数据类型的流水线,而无需使用类型插件。 至关重要的是,它允许完全与 ORM 使用的 select()Row 构造的范式进行交互,这些构造引用了 ORM 映射的类类型(例如包含用户映射实例的 Row,例如我们教程中使用的 UserAddress 示例)。 虽然 Python 类型目前对于元组类型的定制支持非常有限(其中 PEP 646,第一个试图处理类似元组的对象的 pep,在其功能上有意受到了限制,并且本身尚不适用于任意元组操作),但已经设计出了一种相当不错的方法,允许基本的 select() -> Result -> Row 类型功能,包括对 ORM 类的支持,在要将 Row 对象展开为单独的列条目时,添加了一个小的面向类型的访问器,允许各个 Python 值保持与其来源的 SQL 表达式相关联的 Python 类型(翻译:它有效)。

SQL 表达式类型化 - 示例

类型行为的简要介绍。 注释指示了在 vscode 上悬停在代码上会看到什么(或者使用 reveal_type() 助手时,大致会显示什么类型工具):

  • Python 中的简单类型赋给 SQL 表达式
# (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,例如 InsertInsert.returning(),都被打包到一个 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()方法时,传递到ResultRow。为了将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.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-stubssqlalchemy2-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()
注意事项 - 必须卸载所有存根

关于类型支持的一个关键注意事项是 必须卸载所有 SQLAlchemy 存根包 才能使类型化工作。当对 Python 虚拟环境运行 mypy 时,只需卸载这些包。但是,目前 SQLAlchemy 存根包也是 typeshed 的一部分,它本身被捆绑到一些类型工具中,例如 Pylance,因此在某些情况下,可能需要定位这些包的文件并将其删除,以确保它们不会干扰新的类型化工作正常运行。

一旦 SQLAlchemy 2.0 正式发布,typeshed 将从其自己的存根源中删除 SQLAlchemy。

原理与概述

提示

本节是一个架构讨论。要快速查看新的类型,请跳转到 SQL 表达式类型化 - 示例。

sqlalchemy2-stubs 中,SQL 表达式被类型化为 泛型,然后引用了 TypeEngine 对象,例如 IntegerDateTimeString 作为它们的泛型参数(如 Column[Integer])。这本身是对原始 Dropbox sqlalchemy-stubs 包的偏离,其中 Column 及其基本构造直接泛型化为 Python 类型,如 intdatetimestr。希望 Integer / DateTime / String 本身是对 int / datetime / str 泛型化的,就有可能保持两个层次的信息并能够通过 TypeEngine 作为中介构造从列表达式中提取 Python 类型。然而,事实并非如此,因为 PEP 484 实际上没有足够丰富的功能集使得这成为可行的,缺乏诸如 更高种类的类型变量 等能力。

经过对当前 深度评估,SQLAlchemy 2.0 在这一领域实现了 sqlalchemy-stubs 的原始智慧,并直接将列表达式与 Python 类型关联起来。这意味着如果有 SQL 表达式到不同子类型的情况,比如 Column(VARCHAR)Column(Unicode),那么这两个 String 子类型的具体细节并不会随着类型一起传递,但实际上这通常不是问题,通常更有用的是 Python 类型立即出现,因为它代表了直接存储和接收此列的 Python 数据。

具体来说,这意味着像 Column('id', Integer) 这样的表达式被类型化为 Column[int]。这允许建立一个可行的 SQLAlchemy 构造 -> Python 数据类型的管道,而无需使用类型插件。至关重要的是,它允许完全与 ORM 的范式进行互操作,即使用引用 ORM 映射类类型的 select()Row 构造(例如包含用户映射实例的 Row,例如我们教程中使用的 UserAddress 示例)。虽然 Python 类型当前对元组类型的自定义支持非常有限(其中 PEP 646 是第一个试图处理类似元组对象的 pep,但其功能故意受到限制,本身尚不适用于任意元组操作),但已经设计出了一个相当不错的方法,允许基本的 select() -> Result -> Row 类型功能运行,包括对 ORM 类的支持,在将 Row 对象拆包为单独的列条目时,添加了一个小的面向类型的访问器,以便使得每个 Python 值都能保持与其来源的 SQL 表达式关联的 Python 类型(翻译:它可以正常工作)。

SQL 表达式类型 - 示例

对类型行为的简要介绍。注释指示了在 vscode 中悬停在代码上会看到什么(或者在使用 reveal_type() 助手时大致会显示什么类型工具):

  • 将简单的 Python 类型分配给 SQL 表达式
# (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,如 InsertInsert.returning(),都打包成一个 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() 方法时,都会传递到 ResultRow。为了将 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)
  • Core 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-stubssqlalchemy2-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()
要点是 - 必须卸载所有存根

与类型支持相关的一个关键注意事项是必须卸载所有 SQLAlchemy 存根包才能使类型工作。当针对 Python 虚拟环境运行mypy时,只需卸载这些包。然而,SQLAlchemy 存根包目前也是typeshed的一部分,它本身被捆绑到一些类型工具中,如Pylance,因此在某些情况下可能需要定位这些包的文件并删除它们,以确保新的类型正确工作。

一旦 SQLAlchemy 2.0 发布为最终状态,typeshed 将从其自己的存根源中删除 SQLAlchemy。


SqlAlchemy 2.0 中文文档(五十七)(5)https://developer.aliyun.com/article/1563167

相关文章
|
1月前
|
SQL 数据库连接 API
SqlAlchemy 2.0 中文文档(五十五)(3)
SqlAlchemy 2.0 中文文档(五十五)
35 1
|
1月前
|
SQL 缓存 数据库连接
SqlAlchemy 2.0 中文文档(五十五)(1)
SqlAlchemy 2.0 中文文档(五十五)
20 1
|
1月前
|
API 数据库 Python
SqlAlchemy 2.0 中文文档(五十五)(6)
SqlAlchemy 2.0 中文文档(五十五)
16 1
|
1月前
|
SQL 安全 数据库连接
SqlAlchemy 2.0 中文文档(五十五)(5)
SqlAlchemy 2.0 中文文档(五十五)
17 1
|
1月前
|
SQL 缓存 编译器
SqlAlchemy 2.0 中文文档(五十五)(4)
SqlAlchemy 2.0 中文文档(五十五)
22 1
|
1月前
|
SQL 数据库连接 数据库
SqlAlchemy 2.0 中文文档(五十五)(2)
SqlAlchemy 2.0 中文文档(五十五)
21 1
|
1月前
|
SQL 关系型数据库 MySQL
SqlAlchemy 2.0 中文文档(五十七)(7)
SqlAlchemy 2.0 中文文档(五十七)
16 0
|
1月前
|
SQL Python
SqlAlchemy 2.0 中文文档(五十七)(5)
SqlAlchemy 2.0 中文文档(五十七)
11 0
|
1月前
|
SQL API Python
SqlAlchemy 2.0 中文文档(五十七)(6)
SqlAlchemy 2.0 中文文档(五十七)
14 0
|
1月前
|
SQL Oracle 关系型数据库
SqlAlchemy 2.0 中文文档(五十七)(3)
SqlAlchemy 2.0 中文文档(五十七)
15 0