SqlAlchemy 2.0 中文文档(三十一)(1)

简介: SqlAlchemy 2.0 中文文档(三十一)


原文:docs.sqlalchemy.org/en/20/contents.html

声明式扩展

原文:docs.sqlalchemy.org/en/20/orm/extensions/declarative/index.html

声明式映射 API 特定的扩展。

1.4 版本更改:绝大部分声明式扩展现在已整合到 SQLAlchemy ORM 中,并可从 sqlalchemy.orm 命名空间导入。请参阅声明式映射的文档以获取新文档。有关更改的概述,请参阅声明式现已与 ORM 整合,并带有新功能。

对象名称 描述
AbstractConcreteBase 一个用于“具体”声明式映射的辅助类。
ConcreteBase 一个用于“具体”声明式映射的辅助类。
DeferredReflection 一个用于基于延迟反射步骤构建映射的辅助类。
class sqlalchemy.ext.declarative.AbstractConcreteBase

一个用于“具体”声明式映射的辅助类。

AbstractConcreteBase 将自动使用 polymorphic_union() 函数,对所有作为此类的子类映射的表执行。该函数通过 __declare_first__() 函数调用,这实际上是一个 before_configured() 事件的钩子。

AbstractConcreteBase 应用 Mapper 到其直接继承的类,就像对任何其他声明式映射的类一样。然而,Mapper 没有映射到任何特定的 Table 对象。相反,它直接映射到由 polymorphic_union() 产生的“多态”可选择的对象,并且不执行自己的持久化操作。与 ConcreteBase 相比,后者将其直接继承的类映射到直接存储行的实际 Table

注意

AbstractConcreteBase延迟了基类的映射器创建,直到所有子类都已定义,因为它需要创建一个针对包含所有子类表的可选择项的映射。为了实现这一点,它等待映射器配置事件发生,然后扫描所有配置的子类,并设置一个将一次性查询所有子类的映射。

虽然此事件通常会自动调用,但在AbstractConcreteBase的情况下,如果第一个操作是针对此基类的查询,则可能需要在定义所有子类映射之后显式调用它。为此,一旦所有期望的类都已配置,可以调用正在使用的registry上的registry.configure()方法,该方法可在特定声明基类的关系中使用:

Base.registry.configure()

示例:

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.ext.declarative import AbstractConcreteBase
class Base(DeclarativeBase):
    pass
class Employee(AbstractConcreteBase, Base):
    pass
class Manager(Employee):
    __tablename__ = 'manager'
    employee_id = Column(Integer, primary_key=True)
    name = Column(String(50))
    manager_data = Column(String(40))
    __mapper_args__ = {
        'polymorphic_identity':'manager',
        'concrete':True
    }
Base.registry.configure()

抽象基类在声明时以一种特殊的方式处理;在类配置时,它的行为类似于声明式的混入或__abstract__基类。一旦类被配置并生成映射,它会被映射自身,但在其所有子类之后。这是在任何其他 SQLAlchemy API 功能中都找不到的非常独特的映射系统。

使用这种方法,我们可以指定将在映射的子类上发生的列和属性,就像我们通常在 Mixin 和自定义基类中所做的那样:

from sqlalchemy.ext.declarative import AbstractConcreteBase
class Company(Base):
    __tablename__ = 'company'
    id = Column(Integer, primary_key=True)
class Employee(AbstractConcreteBase, Base):
    strict_attrs = True
    employee_id = Column(Integer, primary_key=True)
    @declared_attr
    def company_id(cls):
        return Column(ForeignKey('company.id'))
    @declared_attr
    def company(cls):
        return relationship("Company")
class Manager(Employee):
    __tablename__ = 'manager'
    name = Column(String(50))
    manager_data = Column(String(40))
    __mapper_args__ = {
        'polymorphic_identity':'manager',
        'concrete':True
    }
Base.registry.configure()

然而,当我们使用我们的映射时,ManagerEmployee都将拥有一个可独立使用的.company属性:

session.execute(
    select(Employee).filter(Employee.company.has(id=5))
)

参数:

strict_attrs

当在基类上指定时,“严格”属性模式被启用,试图将基类上的 ORM 映射属性限制为仅当下立即存在的属性,同时仍保留“多态”加载行为。

2.0 版中新增。

另请参阅

ConcreteBase

具体表继承

抽象具体类

类签名

sqlalchemy.ext.declarative.AbstractConcreteBase (sqlalchemy.ext.declarative.extensions.ConcreteBase)

class sqlalchemy.ext.declarative.ConcreteBase

用于‘具体’声明映射的辅助类。

ConcreteBase 会自动使用 polymorphic_union() 函数,针对所有映射为该类的子类的表。该函数通过 __declare_last__() 函数调用,这实质上是 after_configured() 事件的钩子。

ConcreteBase 为类本身生成一个映射表。与 AbstractConcreteBase 相比,后者不会。

示例:

from sqlalchemy.ext.declarative import ConcreteBase
class Employee(ConcreteBase, Base):
    __tablename__ = 'employee'
    employee_id = Column(Integer, primary_key=True)
    name = Column(String(50))
    __mapper_args__ = {
                    'polymorphic_identity':'employee',
                    'concrete':True}
class Manager(Employee):
    __tablename__ = 'manager'
    employee_id = Column(Integer, primary_key=True)
    name = Column(String(50))
    manager_data = Column(String(40))
    __mapper_args__ = {
                    'polymorphic_identity':'manager',
                    'concrete':True}

polymorphic_union() 使用的鉴别器列的默认名称为 type。为了适应映射的用例,其中映射表中的实际列已命名为 type,可以通过设置 _concrete_discriminator_name 属性来配置鉴别器名称:

class Employee(ConcreteBase, Base):
    _concrete_discriminator_name = '_concrete_discriminator'

自版本 1.3.19 中新增:为 ConcreteBase 添加了 _concrete_discriminator_name 属性,以便自定义虚拟鉴别器列名称。

自版本 1.4.2 中更改:只需将 _concrete_discriminator_name 属性放置在最基类上即可使所有子类正确生效。如果映射列名称与鉴别器名称冲突,则现在会显示显式错误消息,而在 1.3.x 系列中会有一些警告,然后生成一个无用的查询。

另请参阅

AbstractConcreteBase

具体表继承

class sqlalchemy.ext.declarative.DeferredReflection

一个用于基于延迟反射步骤构建映射的辅助类。

通常情况下,通过将一个 Table 对象设置为具有 autoload_with=engine 的 __table__ 属性,可以使用反射来使用声明。一个声明性类。需要注意的是,在构建普通声明性映射的时候,Table 必须是完全反映的,或者至少有一个主键列,这意味着在类声明时必须可用 Engine

DeferredReflection mixin 将映射器的构建移动到稍后的时间点,在调用首先反射到目前为止创建的所有 Table 对象的特定方法之后。类可以定义如下:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import DeferredReflection
Base = declarative_base()
class MyClass(DeferredReflection, Base):
    __tablename__ = 'mytable'

在上面,MyClass 还没有映射。在上述方式定义了一系列类之后,可以使用 prepare() 反射所有表并创建映射:

engine = create_engine("someengine://...")
DeferredReflection.prepare(engine)

DeferredReflection mixin 可以应用于单个类,用作声明基类本身,或用于自定义抽象类。使用抽象基类允许仅为特定准备步骤准备一部分类,这对于使用多个引擎的应用程序是必要的。例如,如果一个应用程序有两个引擎,您可能会使用两个基类,并分别准备每个基类,例如:

class ReflectedOne(DeferredReflection, Base):
    __abstract__ = True
class ReflectedTwo(DeferredReflection, Base):
    __abstract__ = True
class MyClass(ReflectedOne):
    __tablename__ = 'mytable'
class MyOtherClass(ReflectedOne):
    __tablename__ = 'myothertable'
class YetAnotherClass(ReflectedTwo):
    __tablename__ = 'yetanothertable'
# ... etc.

在上面,ReflectedOneReflectedTwo 的类层次结构可以分别配置:

ReflectedOne.prepare(engine_one)
ReflectedTwo.prepare(engine_two)

成员

prepare()

另请参阅

使用 DeferredReflection - 在 使用声明式配置表 部分。

classmethod prepare(bind: Engine | Connection, **reflect_kw: Any) → None

反射所有当前 DeferredReflection 子类的所有 Table 对象

参数:

  • bind
    EngineConnection 实例
    …versionchanged:: 2.0.16 现在也接受 Connection
  • **reflect_kw
    传递给 MetaData.reflect() 的其他关键字参数,例如 MetaData.reflect.views
    新版本 2.0.16 中的内容。

Mypy / Pep-484 对 ORM 映射的支持

原文:docs.sqlalchemy.org/en/20/orm/extensions/mypy.html

当使用直接引用 Column 对象而不是 SQLAlchemy 2.0 中引入的 mapped_column() 构造时,支持 PEP 484 类型注释以及 MyPy 类型检查工具。

自 2.0 版开始已被弃用:SQLAlchemy Mypy 插件已弃用,并且可能在 SQLAlchemy 2.1 发布时被移除。我们建议用户尽快迁移。

无法跨不断变化的 mypy 发布维护此插件,未来的稳定性不能保证。

现代 SQLAlchemy 现在提供了 完全符合 pep-484 的映射语法;请参阅链接的部分以获取迁移详情。

安装

仅适用于 SQLAlchemy 2.0:不应安装存根,并且应完全卸载诸如 sqlalchemy-stubssqlalchemy2-stubs 等软件包。

Mypy 包本身是一个依赖项。

可以使用 pip 使用“mypy”额外钩子安装 Mypy:

pip install sqlalchemy[mypy]

插件本身如 Configuring mypy to use Plugins 中描述的那样配置,使用 sqlalchemy.ext.mypy.plugin 模块名,例如在 setup.cfg 中:

[mypy]
plugins = sqlalchemy.ext.mypy.plugin

插件功能

Mypy 插件的主要目的是拦截并修改 SQLAlchemy 声明性映射 的静态定义,使其与它们在被其 Mapper 对象 instrumented 后的结构相匹配。这允许类结构本身以及使用类的代码对 Mypy 工具有意义,否则基于当前声明性映射的功能,这是不可能的。该插件类似于需要为类似 dataclasses 这样的库修改类的动态插件。

为了涵盖这种情况经常发生的主要区域,考虑以下 ORM 映射,使用 User 类的典型示例:

from sqlalchemy import Column, Integer, String, select
from sqlalchemy.orm import declarative_base
# "Base" is a class that is created dynamically from the
# declarative_base() function
Base = declarative_base()
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
# "some_user" is an instance of the User class, which
# accepts "id" and "name" kwargs based on the mapping
some_user = User(id=5, name="user")
# it has an attribute called .name that's a string
print(f"Username: {some_user.name}")
# a select() construct makes use of SQL expressions derived from the
# User class itself
select_stmt = select(User).where(User.id.in_([3, 4, 5])).where(User.name.contains("s"))

上述,Mypy 扩展可以执行的步骤包括:

  • 解释由 declarative_base() 生成的 Base 动态类,以便从中继承的类被认为是映射的。它还可以适应在使用装饰器进行声明式映射(无声明式基类)中描述的类装饰器方法。
  • 对在声明式“内联”样式中定义的 ORM 映射属性进行类型推断,例如上面示例中 User 类的 idname 属性。这包括 User 的实例将使用 int 类型的 idstr 类型的 name。还包括当访问 User.idUser.name 类级属性时,如上面的 select() 语句中所示,它们与 SQL 表达式行为兼容,这是从 InstrumentedAttribute 属性描述符类派生的。
  • __init__() 方法应用于尚未包含显式构造函数的映射类,该构造函数接受检测到的所有映射属性的特定类型的关键字参数。

当 Mypy 插件处理上述文件时,传递给 Mypy 工具的结果静态类定义和 Python 代码等效于以下内容:

from sqlalchemy import Column, Integer, String, select
from sqlalchemy.orm import Mapped
from sqlalchemy.orm.decl_api import DeclarativeMeta
class Base(metaclass=DeclarativeMeta):
    __abstract__ = True
class User(Base):
    __tablename__ = "user"
    id: Mapped[Optional[int]] = Mapped._special_method(
        Column(Integer, primary_key=True)
    )
    name: Mapped[Optional[str]] = Mapped._special_method(Column(String))
    def __init__(self, id: Optional[int] = ..., name: Optional[str] = ...) -> None: ...
some_user = User(id=5, name="user")
print(f"Username: {some_user.name}")
select_stmt = select(User).where(User.id.in_([3, 4, 5])).where(User.name.contains("s"))

上述已经采取的关键步骤包括:

  • Base 类现在明确地是基于 DeclarativeMeta 类定义的,而不再是一个动态类。
  • idname 属性是基于 Mapped 类定义的,该类代表一个在类和实例级别表现出不同行为的 Python 描述符。Mapped 类现在是用于所有 ORM 映射属性的 InstrumentedAttribute 类的基类。
    Mapped 被定义为一个针对任意 Python 类型的通用类,这意味着特定的 Mapped 实例与特定的 Python 类型相关联,例如上面的 Mapped[Optional[int]]Mapped[Optional[str]
  • 声明性映射属性赋值的右侧被移除,因为这类似于Mapper类通常要执行的操作,即它将用InstrumentedAttribute](…/internals.html#sqlalchemy.orm.InstrumentedAttribute   “sqlalchemy.orm.InstrumentedAttribute”)的特定实例替换这些属性。原始表达式移动到一个函数调用中,这样可以仍然进行类型检查而不与表达式的左侧冲突。对于  Mypy 来说,左侧的类型注释足以理解属性的行为。
  • 添加了User.__init__()方法的类型存根,其中包括了正确的关键字和数据类型。

用法

以下各小节将讨论到目前为止已经考虑到的符合 PEP-484 的各种使用情况。

基于 TypeEngine 的列的内省

对于包含显式数据类型的映射列,当它们被映射为内联属性时,映射类型将被自动内省:

class MyClass(Base):
    # ...
    id = Column(Integer, primary_key=True)
    name = Column("employee_name", String(50), nullable=False)
    other_name = Column(String(50))

上述,idnameother_name的最终类级数据类型将被内省为Mapped[Optional[int]]Mapped[Optional[str]]Mapped[Optional[str]]。这些类型默认始终被认为是Optional,即使对于主键和非空列也是如此。原因是因为虽然数据库列idname不能为 NULL,但 Python 属性idname很可能是None,而不需要显式的构造函数:

>>> m1 = MyClass()
>>> m1.id
None

上述列的类型可以被显式地声明,提供了更清晰的自我文档化以及能够控制哪些类型是可选的两个优点:

class MyClass(Base):
    # ...
    id: int = Column(Integer, primary_key=True)
    name: str = Column("employee_name", String(50), nullable=False)
    other_name: Optional[str] = Column(String(50))

Mypy 插件将接受上述intstrOptional[str]并将它们转换为包含在其周围的Mapped[]类型。Mapped[]构造也可以被显式使用:

from sqlalchemy.orm import Mapped
class MyClass(Base):
    # ...
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column("employee_name", String(50), nullable=False)
    other_name: Mapped[Optional[str]] = Column(String(50))

当类型是非可选时,这意味着从MyClass的实例中访问的属性将被认为是非None的:

mc = MyClass(...)
# will pass mypy --strict
name: str = mc.name

对于可选属性,Mypy 认为类型必须包含 None,否则就是Optional

mc = MyClass(...)
# will pass mypy --strict
other_name: Optional[str] = mc.name

无论映射的属性是否被标记为Optional__init__()方法的生成都仍然认为所有关键字都是可选的。这再次与 SQLAlchemy ORM 在创建构造函数时实际执行的操作相匹配,不应与诸如 Python dataclasses之类的验证系统的行为混淆,后者将生成一个根据注释匹配的构造函数,包括可选和必需的属性。

没有明确类型的列

包含ForeignKey修饰符的列在 SQLAlchemy 声明映射中不需要指定数据类型。对于这种类型的属性,Mypy 插件将通知用户需要发送明确的类型:

# .. other imports
from sqlalchemy.sql.schema import ForeignKey
Base = declarative_base()
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    user_id = Column(ForeignKey("user.id"))

插件将按以下方式传递消息:

$ mypy test3.py --strict
test3.py:20: error: [SQLAlchemy Mypy plugin] Can't infer type from
ORM mapped expression assigned to attribute 'user_id'; please specify a
Python type or Mapped[<python type>] on the left hand side.
Found 1 error in 1 file (checked 1 source file)

要解决问题,请对Address.user_id列应用明确的类型注释:

class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    user_id: int = Column(ForeignKey("user.id"))

使用命令式表映射列

在命令式表样式中,Column定义位于与映射属性本身分开的Table构造内。Mypy 插件不考虑这个Table,而是支持可以明确声明属性,并且必须使用Mapped类将其标识为映射属性:

class MyClass(Base):
    __table__ = Table(
        "mytable",
        Base.metadata,
        Column(Integer, primary_key=True),
        Column("employee_name", String(50), nullable=False),
        Column(String(50)),
    )
    id: Mapped[int]
    name: Mapped[str]
    other_name: Mapped[Optional[str]]

上述Mapped注释被视为映射列,并将包含在默认构造函数中,同时为MyClass在类级别和实例级别提供正确的类型配置文件。

映射关系

该插件对使用类型推断来检测关系类型有限支持。对于所有无法检测类型的情况,它将发出信息丰富的错误消息,并且在所有情况下,可以明确提供适当的类型,要么使用Mapped类,要么选择在内联声明中省略它。插件还需要确定关系是指向集合还是标量,并且为此依赖于relationship.uselist和/或relationship.collection_class参数的显式值。如果这些参数都不存在,则需要明确的类型,以及如果relationship()的目标类型是字符串或可调用对象,而不是类:

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    user_id: int = Column(ForeignKey("user.id"))
    user = relationship(User)

上述映射将产生以下错误:

test3.py:22: error: [SQLAlchemy Mypy plugin] Can't infer scalar or
collection for ORM mapped expression assigned to attribute 'user'
if both 'uselist' and 'collection_class' arguments are absent from the
relationship(); please specify a type annotation on the left hand side.
Found 1 error in 1 file (checked 1 source file)

可以通过使用relationship(User, uselist=False)或提供类型来解决错误,在这种情况下是标量User对象:

class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    user_id: int = Column(ForeignKey("user.id"))
    user: User = relationship(User)

对于集合,类似的模式也适用,即在没有uselist=Truerelationship.collection_class的情况下,可以使用诸如List之类的集合注释。还可以完全适当地使用类的字符串名称进行注释,如 pep-484 所支持,确保根据需要在TYPE_CHECKING 块中导入类:

from typing import TYPE_CHECKING, List
from .mymodel import Base
if TYPE_CHECKING:
    # if the target of the relationship is in another module
    # that cannot normally be imported at runtime
    from .myaddressmodel import Address
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    addresses: List["Address"] = relationship("Address")

与列一样,Mapped 类也可以显式应用:

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    addresses: Mapped[List["Address"]] = relationship("Address", back_populates="user")
class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    user_id: int = Column(ForeignKey("user.id"))
    user: Mapped[User] = relationship(User, back_populates="addresses")


SqlAlchemy 2.0 中文文档(三十一)(2)https://developer.aliyun.com/article/1562898

相关文章
|
6月前
|
SQL 数据库 Python
SqlAlchemy 2.0 中文文档(三十一)(3)
SqlAlchemy 2.0 中文文档(三十一)
35 1
|
6月前
|
JSON 测试技术 数据格式
SqlAlchemy 2.0 中文文档(三十一)(4)
SqlAlchemy 2.0 中文文档(三十一)
45 1
|
6月前
|
SQL 测试技术 数据库
SqlAlchemy 2.0 中文文档(三十一)(2)
SqlAlchemy 2.0 中文文档(三十一)
47 1
|
6月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档(十九)(2)
SqlAlchemy 2.0 中文文档(十九)
46 2
|
6月前
|
SQL 存储 关系型数据库
SqlAlchemy 2.0 中文文档(三十四)(4)
SqlAlchemy 2.0 中文文档(三十四)
51 1
|
6月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(三十四)(5)
SqlAlchemy 2.0 中文文档(三十四)
50 0
|
6月前
|
SQL Java Go
SqlAlchemy 2.0 中文文档(十九)(1)
SqlAlchemy 2.0 中文文档(十九)
48 1
|
6月前
|
SQL API 数据安全/隐私保护
SqlAlchemy 2.0 中文文档(三十二)(3)
SqlAlchemy 2.0 中文文档(三十二)
43 1
|
6月前
|
JSON 测试技术 数据格式
SqlAlchemy 2.0 中文文档(三十一)(5)
SqlAlchemy 2.0 中文文档(三十一)
51 0
|
6月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(三十四)(2)
SqlAlchemy 2.0 中文文档(三十四)
67 0

热门文章

最新文章