使用声明式配置表

正如在 声明式映射 中介绍的那样,声明式风格包括同时生成映射的 Table 对象的能力,或者直接容纳 Table 或其他 FromClause 对象。

以下示例假定声明式基类如下

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

以下所有示例都说明了一个类继承自上述 Base。在 使用装饰器的声明式映射 (无声明式基类) 中介绍的装饰器风格也完全支持以下所有示例,以及包括由 declarative_base() 生成的基类在内的旧式声明式基类形式。

使用 mapped_column() 的声明式表

当使用声明式时,在大多数情况下,要映射的类的正文包括一个属性 __tablename__,该属性指示应与映射一起生成的 Table 的字符串名称。mapped_column() 构造在普通的 Column 类中不具备额外的 ORM 特定配置功能,然后在类主体中使用,以指示表中的列。下面的示例说明了在声明式映射中此构造最基本的用法

from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(50), nullable=False)
    fullname = mapped_column(String)
    nickname = mapped_column(String(30))

在上面,mapped_column() 构造以内联方式放置在类定义中作为类级别属性。在声明类时,声明式映射过程将针对与声明式 Base 关联的 MetaData 集合生成一个新的 Table 对象;mapped_column() 的每个实例都将用于在此过程中生成一个 Column 对象,该对象将成为此 Table 对象的 Table.columns 集合的一部分。

在上面的示例中,声明式将构建一个 Table 构造,它等效于以下内容

# equivalent Table object produced
user_table = Table(
    "user",
    Base.metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("fullname", String()),
    Column("nickname", String(30)),
)

当映射上面的 User 类时,可以通过 __table__ 属性直接访问此 Table 对象;这在 访问表和元数据 中进一步描述。

mapped_column() 构造接受 Column 构造接受的所有参数,以及额外的 ORM 特定参数。mapped_column.__name 字段指示数据库列的名称,通常省略,因为声明式过程将使用赋予构造的属性名称并将其分配为列的名称 (在上面的示例中,这指的是名称 id, name, fullname, nickname)。分配备用 mapped_column.__name 也有效,其中生成的 Column 将在 SQL 和 DDL 语句中使用给定的名称,而 User 映射类将继续允许使用给定的属性名称访问该属性,而与赋予列本身的名称无关 (更多信息请参见 显式命名声明式映射列)。

提示

mapped_column() 构造仅在声明式类映射中有效。当使用 Core 构建 Table 对象以及当使用 命令式表 配置时,仍然需要 Column 构造以指示数据库列的存在。

另请参阅

映射表列 - 包含关于影响 Mapper 如何解释传入的 Column 对象的其他说明。

使用注解声明式表 (用于 mapped_column() 的类型注解形式)

mapped_column() 构造能够从与在声明式映射类中声明的属性关联的 PEP 484 类型注解中派生其列配置信息。这些类型注解 (如果使用) 必须 存在于名为 Mapped 的特殊 SQLAlchemy 类型中,Mapped 是一种 泛型 类型,然后在其中指示特定的 Python 类型。

下面说明了上一节中的映射,添加了 Mapped 的使用

from typing import Optional

from sqlalchemy import String
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"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    fullname: Mapped[Optional[str]]
    nickname: Mapped[Optional[str]] = mapped_column(String(30))

在上面,当声明式处理每个类属性时,如果存在,每个 mapped_column() 将从左侧相应的 Mapped 类型注解中派生其他参数。此外,当遇到没有为属性赋值的 Mapped 类型注解时,声明式将隐式生成一个空的 mapped_column() 指令 (此形式的灵感来自 Python dataclasses 中使用的类似风格);此 mapped_column() 构造继续从存在的 Mapped 注解中派生其配置。

mapped_column()Mapped 注解中派生数据类型和可空性

mapped_column()Mapped 注解中派生的两个特性是

  • 数据类型 - Mapped 内部给定的 Python 类型,如果存在于 typing.Optional 构造中,则与 TypeEngine 子类 (例如 Integer, String, DateTimeUuid,仅举几个常见的类型) 相关联。

    数据类型是基于 Python 类型到 SQLAlchemy 数据类型的字典确定的。此字典是完全可自定义的,如下一节 自定义类型映射 中详述。默认类型映射在下面的代码示例中实现

    from typing import Any
    from typing import Dict
    from typing import Type
    
    import datetime
    import decimal
    import uuid
    
    from sqlalchemy import types
    
    # default type mapping, deriving the type for mapped_column()
    # from a Mapped[] annotation
    type_map: Dict[Type[Any], TypeEngine[Any]] = {
        bool: types.Boolean(),
        bytes: types.LargeBinary(),
        datetime.date: types.Date(),
        datetime.datetime: types.DateTime(),
        datetime.time: types.Time(),
        datetime.timedelta: types.Interval(),
        decimal.Decimal: types.Numeric(),
        float: types.Float(),
        int: types.Integer(),
        str: types.String(),
        uuid.UUID: types.Uuid(),
    }

    如果 mapped_column() 构造指示了作为传递给 mapped_column.__type 参数的显式类型,则给定的 Python 类型将被忽略。

  • 可空性 - mapped_column() 构造将首先通过 mapped_column.nullable 参数 (传递为 TrueFalse) 指示其 ColumnNULLNOT NULL。此外,如果存在 mapped_column.primary_key 参数并设置为 True,则这也将暗示该列应为 NOT NULL

    两者都不存在的情况下,Mapped 类型注解中 typing.Optional[] 的存在将用于确定可空性,其中 typing.Optional[] 表示 NULL,而 typing.Optional[] 的缺失表示 NOT NULL。如果根本没有 Mapped[] 注解,并且没有 mapped_column.nullablemapped_column.primary_key 参数,则使用 SQLAlchemy 通常的 Column 默认值 NULL

    在下面的示例中,iddata 列将为 NOT NULL,而 additional_info 列将为 NULL

    from typing import Optional
    
    from sqlalchemy.orm import DeclarativeBase
    from sqlalchemy.orm import Mapped
    from sqlalchemy.orm import mapped_column
    
    
    class Base(DeclarativeBase):
        pass
    
    
    class SomeClass(Base):
        __tablename__ = "some_table"
    
        # primary_key=True, therefore will be NOT NULL
        id: Mapped[int] = mapped_column(primary_key=True)
    
        # not Optional[], therefore will be NOT NULL
        data: Mapped[str]
    
        # Optional[], therefore will be NULL
        additional_info: Mapped[Optional[str]]

    拥有一个 mapped_column(),其可空性与注解暗示的可空性不同也是完全有效的。例如,ORM 映射属性可以注解为允许在 Python 代码中使用 None,因为它最初正在创建和填充对象,但是该值最终将被写入 NOT NULL 的数据库列。mapped_column.nullable 参数 (如果存在) 将始终优先

    class SomeClass(Base):
        # ...
    
        # will be String() NOT NULL, but can be None in Python
        data: Mapped[Optional[str]] = mapped_column(nullable=False)

    类似地,对于由于某种原因需要在模式级别为 NULL 的数据库列写入的非 None 属性,可以将 mapped_column.nullable 设置为 True

    class SomeClass(Base):
        # ...
    
        # will be String() NULL, but type checker will not expect
        # the attribute to be None
        data: Mapped[str] = mapped_column(nullable=True)

自定义类型映射

上一节中描述的 Python 类型到 SQLAlchemy TypeEngine 类型的映射默认设置为 sqlalchemy.sql.sqltypes 模块中存在的硬编码字典。但是,协调声明式映射过程的 registry 对象将首先查阅本地用户定义的类型字典,该字典可以作为构造 registry 时的 registry.type_annotation_map 参数传递,该字典可以与首次使用时的 DeclarativeBase 超类关联。

例如,如果我们希望对 int 使用 BIGINT 数据类型,对 datetime.datetime 使用 timezone=TrueTIMESTAMP 数据类型,并且仅在 Microsoft SQL Server 上,当我们使用 Python str 时,我们才想使用 NVARCHAR 数据类型,则可以如下配置注册表和声明式基类

import datetime

from sqlalchemy import BIGINT, NVARCHAR, String, TIMESTAMP
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    type_annotation_map = {
        int: BIGINT,
        datetime.datetime: TIMESTAMP(timezone=True),
        str: String().with_variant(NVARCHAR, "mssql"),
    }


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    date: Mapped[datetime.datetime]
    status: Mapped[str]

下面说明了为上述映射生成的 CREATE TABLE 语句,首先在 Microsoft SQL Server 后端,说明了 NVARCHAR 数据类型

>>> from sqlalchemy.schema import CreateTable
>>> from sqlalchemy.dialects import mssql, postgresql
>>> print(CreateTable(SomeClass.__table__).compile(dialect=mssql.dialect()))
CREATE TABLE some_table ( id BIGINT NOT NULL IDENTITY, date TIMESTAMP NOT NULL, status NVARCHAR(max) NOT NULL, PRIMARY KEY (id) )

然后在 PostgreSQL 后端,说明了 TIMESTAMP WITH TIME ZONE

>>> print(CreateTable(SomeClass.__table__).compile(dialect=postgresql.dialect()))
CREATE TABLE some_table ( id BIGSERIAL NOT NULL, date TIMESTAMP WITH TIME ZONE NOT NULL, status VARCHAR NOT NULL, PRIMARY KEY (id) )

通过使用诸如 TypeEngine.with_variant() 之类的方法,我们能够构建一个针对不同后端的需求进行自定义的类型映射,同时仍然能够使用简洁的仅注解 mapped_column() 配置。除了这一点之外,还有另外两个级别的 Python 类型可配置性可用,在接下来的两节中描述。

类型映射内部的 Union 类型

在版本 2.0.37 中更改: 本节中描述的功能已修复和增强,以保持一致的工作。在此更改之前,type_annotation_map 中支持 union 类型,但是该功能在 union 语法之间以及在如何处理 None 方面表现出不一致的行为。在尝试使用本节中描述的功能之前,请确保 SQLAlchemy 是最新的。

SQLAlchemy 支持在 type_annotation_map 中映射 union 类型,以允许映射可以支持多种 Python 类型的数据库类型,例如 JSONJSONB

from typing import Union
from sqlalchemy import JSON
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.schema import CreateTable

# new style Union using a pipe operator
json_list = list[int] | list[str]

# old style Union using Union explicitly
json_scalar = Union[float, str, bool]


class Base(DeclarativeBase):
    type_annotation_map = {
        json_list: postgresql.JSONB,
        json_scalar: JSON,
    }


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    list_col: Mapped[list[str] | list[int]]

    # uses JSON
    scalar_col: Mapped[json_scalar]

    # uses JSON and is also nullable=True
    scalar_col_nullable: Mapped[json_scalar | None]

    # these forms all use JSON as well due to the json_scalar entry
    scalar_col_newstyle: Mapped[float | str | bool]
    scalar_col_oldstyle: Mapped[Union[float, str, bool]]
    scalar_col_mixedstyle: Mapped[Optional[float | str | bool]]

上面的示例将 list[int]list[str] 的 union 映射到 Postgresql JSONB 数据类型,同时命名 float, str, bool 的 union 将匹配到 JSON 数据类型。在 Mapped 构造中声明的等效 union 将匹配到类型映射中的相应条目。

union 类型的匹配基于 union 的内容,而不管各个类型如何命名,并且另外排除使用 None 类型。也就是说,json_scalar 也将匹配到 str | bool | float | None。它不会 匹配到作为此 union 的子集或超集的 union;也就是说,str | bool 不会匹配,str | bool | float | int 也不会匹配。union 的各个内容 (不包括 None) 必须完全匹配。

None 值在从 type_annotation_map 匹配到 Mapped 时永远不重要,但是作为 Column 的可空性的指示符很重要。当 NoneMapped 构造的形式存在于 union 中时。当存在于 Mapped 中时,它指示在没有更具体的指示符的情况下,Column 将是可空的。此逻辑的工作方式与在 mapped_column() 从 Mapped 注解派生数据类型和可空性 中描述的指示 Optional 类型的方式相同。

上面映射的 CREATE TABLE 语句将如下所示

>>> print(CreateTable(SomeClass.__table__).compile(dialect=postgresql.dialect()))
CREATE TABLE some_table ( id SERIAL NOT NULL, list_col JSONB NOT NULL, scalar_col JSON, scalar_col_not_null JSON NOT NULL, PRIMARY KEY (id) )

虽然联合类型使用“松散”匹配方式,该方式匹配任何等效的子类型集合,但 Python 类型提示也提供了一种创建“类型别名”的方法,这些别名被视为不同的类型,与包含相同组成的另一种类型不等效。下一节 对类型别名类型(由 PEP 695 定义)和 NewType 的支持 中描述了这些类型与 type_annotation_map 的集成。

对类型别名类型(由 PEP 695 定义)和 NewType 的支持

类型映射内部的联合类型 中描述的类型查找相反,Python 类型提示还包括两种以更正式的方式创建组合类型的方法,即使用 typing.NewType 以及 PEP 695 中引入的 type 关键字。这些类型的行为与普通的类型别名(即,将类型分配给变量名)不同,SQLAlchemy 在从类型映射解析这些类型时会遵循这种差异。

在版本 2.0.37 中更改: 本节中描述的 typing.NewType 以及 PEP 695 type 的行为已得到正式确定和纠正。现在,对于在某些 2.0 版本中有效,但将在 SQLAlchemy 2.1 中删除的“松散匹配”模式,将发出弃用警告。请确保 SQLAlchemy 是最新版本,然后再尝试使用本节中描述的功能。

typing 模块允许使用 typing.NewType 创建“新类型”

from typing import NewType

nstr30 = NewType("nstr30", str)
nstr50 = NewType("nstr50", str)

此外,在 Python 3.12 中,引入了由 PEP 695 定义的新功能,该功能提供了 type 关键字来完成类似的任务;使用 type 生成的对象在许多方面与 typing.NewType 相似,在内部被称为 typing.TypeAliasType

type SmallInt = int
type BigInt = int
type JsonScalar = str | float | bool | None

对于 SQLAlchemy 如何处理这些类型对象,以便在 Mapped 内部用于 SQL 类型查找的目的而言,重要的是要注意 Python 不认为两个等效的 typing.TypeAliasTypetyping.NewType 对象相等

# two typing.NewType objects are not equal even if they are both str
>>> nstr50 == nstr30
False

# two TypeAliasType objects are not equal even if they are both int
>>> SmallInt == BigInt
False

# an equivalent union is not equal to JsonScalar
>>> JsonScalar == str | float | bool | None
False

这与普通联合类型的比较行为相反,并告知 SQLAlchemy 的 type_annotation_map 的正确行为。当使用 typing.NewTypePEP 695 type 对象时,类型对象预计在 type_annotation_map 中是显式的,以便从 Mapped 类型中匹配,其中必须声明相同的对象才能进行匹配(不包括 Mapped 内部的类型是否也联合了 None)。这与 类型映射内部的联合类型 中描述的行为不同,在 类型映射内部的联合类型 中,直接引用的普通 Union 将基于 type_annotation_map 中特定类型的组成而不是对象标识来匹配到其他 Unions

在下面的示例中,nstr30nstr50SmallIntBigIntJsonScalar 的组合类型彼此之间没有重叠,并且可以在每个 Mapped 构造中进行不同的命名,并且在 type_annotation_map 中也是显式的。这些类型中的任何一种也可以与 None 联合,或者声明为 Optional[],而不会影响查找,只会派生列的可空性

from typing import NewType

from sqlalchemy import SmallInteger, BigInteger, JSON, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.schema import CreateTable

nstr30 = NewType("nstr30", str)
nstr50 = NewType("nstr50", str)
type SmallInt = int
type BigInt = int
type JsonScalar = str | float | bool | None


class TABase(DeclarativeBase):
    type_annotation_map = {
        nstr30: String(30),
        nstr50: String(50),
        SmallInt: SmallInteger,
        BigInteger: BigInteger,
        JsonScalar: JSON,
    }


class SomeClass(TABase):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    normal_str: Mapped[str]

    short_str: Mapped[nstr30]
    long_str_nullable: Mapped[nstr50 | None]

    small_int: Mapped[SmallInt]
    big_int: Mapped[BigInteger]
    scalar_col: Mapped[JsonScalar]

上面映射的 CREATE TABLE 将说明我们配置的不同整数和字符串变体,如下所示

>>> print(CreateTable(SomeClass.__table__))
CREATE TABLE some_table ( id INTEGER NOT NULL, normal_str VARCHAR NOT NULL, short_str VARCHAR(30) NOT NULL, long_str_nullable VARCHAR(50), small_int SMALLINT NOT NULL, big_int BIGINT NOT NULL, scalar_col JSON, PRIMARY KEY (id) )

关于可空性,JsonScalar 类型在其定义中包含 None,这表示可为空的列。同样,long_str_nullable 列将 None 的联合应用于 nstr50,这与 type_annotation_map 中的 nstr50 类型匹配,同时还将可空性应用于映射的列。其他列都保持 NOT NULL,因为它们未被指示为可选。

将多种类型配置映射到 Python 类型

由于可以使用 registry.type_annotation_map 参数将单个 Python 类型与 TypeEngine 的任何变体配置关联起来,因此另一个功能是能够基于附加的类型限定符将单个 Python 类型与 SQL 类型的不同变体关联起来。一个典型的例子是将 Python str 数据类型映射到不同长度的 VARCHAR SQL 类型。另一个例子是将不同种类的 decimal.Decimal 映射到不同大小的 NUMERIC 列。

Python 的类型提示系统提供了一种很好的方法,可以通过使用 PEP 593 Annotated 泛型类型,为 Python 类型添加额外的元数据,该泛型类型允许将附加信息与 Python 类型捆绑在一起。mapped_column() 构造将通过标识正确地解释 Annotated 对象,以便在 registry.type_annotation_map 中解析它,如下面的示例所示,我们在其中声明了 StringNumeric 的两个变体

from decimal import Decimal

from typing_extensions import Annotated

from sqlalchemy import Numeric
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import registry

str_30 = Annotated[str, 30]
str_50 = Annotated[str, 50]
num_12_4 = Annotated[Decimal, 12]
num_6_2 = Annotated[Decimal, 6]


class Base(DeclarativeBase):
    registry = registry(
        type_annotation_map={
            str_30: String(30),
            str_50: String(50),
            num_12_4: Numeric(12, 4),
            num_6_2: Numeric(6, 2),
        }
    )

传递给 Annotated 容器的 Python 类型(在上面的示例中为 strDecimal 类型)仅对类型提示工具有好处;就 mapped_column() 构造而言,它只需要在 registry.type_annotation_map 字典中执行每个类型对象的查找,而实际上并不需要查看 Annotated 对象内部,至少在这种特定上下文中是这样。同样,传递给 Annotated 的参数(超出底层 Python 类型本身)也不重要,重要的是至少必须存在一个参数,才能使 Annotated 构造有效。然后,我们可以在映射中直接使用这些增强的类型,在映射中,它们将与更具体的类型构造匹配,如下面的示例所示

class SomeClass(Base):
    __tablename__ = "some_table"

    short_name: Mapped[str_30] = mapped_column(primary_key=True)
    long_name: Mapped[str_50]
    num_value: Mapped[num_12_4]
    short_num_value: Mapped[num_6_2]

上面映射的 CREATE TABLE 将说明我们配置的不同 VARCHARNUMERIC 变体,如下所示

>>> from sqlalchemy.schema import CreateTable
>>> print(CreateTable(SomeClass.__table__))
CREATE TABLE some_table ( short_name VARCHAR(30) NOT NULL, long_name VARCHAR(50) NOT NULL, num_value NUMERIC(12, 4) NOT NULL, short_num_value NUMERIC(6, 2) NOT NULL, PRIMARY KEY (short_name) )

虽然将 Annotated 类型链接到不同的 SQL 类型的多样性为我们提供了很大的灵活性,但下一节将说明使用 Declarative 的 Annotated 的第二种方式,这种方式甚至更开放。

将整个列声明映射到 Python 类型

上一节说明了如何使用 PEP 593 Annotated 类型实例作为 registry.type_annotation_map 字典中的键。在这种形式下,mapped_column() 构造实际上并不查看 Annotated 对象本身,它只是用作字典键。但是,Declarative 还能够直接从 Annotated 对象中提取整个预先建立的 mapped_column() 构造。使用这种形式,我们不仅可以定义链接到 Python 类型的不同种类的 SQL 数据类型,而无需使用 registry.type_annotation_map 字典,还可以以可重用的方式设置任意数量的参数,例如可空性、列默认值和约束。

ORM 模型集通常具有某种类型的对所有映射类通用的主键样式。也可能存在常见的列配置,例如带有默认值的 timestamps 和其他预先建立大小和配置的字段。我们可以将这些配置组合到 mapped_column() 实例中,然后将这些实例直接捆绑到 Annotated 实例中,这些实例然后在任意数量的类声明中重复使用。当以这种方式提供 Annotated 对象时,Declarative 将解包该对象,跳过任何不适用于 SQLAlchemy 的其他指令,并且仅搜索 SQLAlchemy ORM 构造。

下面的示例说明了以这种方式使用的各种预配置字段类型,其中我们定义了 intpk,它表示 Integer 主键列,timestamp,它表示 DateTime 类型,该类型将使用 CURRENT_TIMESTAMP 作为 DDL 级别的列默认值,以及 required_name,它是长度为 30 的 String,并且是 NOT NULL

import datetime

from typing_extensions import Annotated

from sqlalchemy import func
from sqlalchemy import String
from sqlalchemy.orm import mapped_column


intpk = Annotated[int, mapped_column(primary_key=True)]
timestamp = Annotated[
    datetime.datetime,
    mapped_column(nullable=False, server_default=func.CURRENT_TIMESTAMP()),
]
required_name = Annotated[str, mapped_column(String(30), nullable=False)]

然后,可以将上面的 Annotated 对象直接在 Mapped 中使用,其中预配置的 mapped_column() 构造将被提取并复制到新的实例,该实例将特定于每个属性

class Base(DeclarativeBase):
    pass


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[intpk]
    name: Mapped[required_name]
    created_at: Mapped[timestamp]

我们上面映射的 CREATE TABLE 看起来像

>>> from sqlalchemy.schema import CreateTable
>>> print(CreateTable(SomeClass.__table__))
CREATE TABLE some_table ( id INTEGER NOT NULL, name VARCHAR(30) NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, PRIMARY KEY (id) )

当以这种方式使用 Annotated 类型时,类型的配置也可能在每个属性的基础上受到影响。对于上面示例中明确使用 mapped_column.nullable 的类型,我们可以将 Optional[] 泛型修饰符应用于我们的任何类型,以便该字段在 Python 级别是可选的或不是可选的,这与数据库中发生的 NULL / NOT NULL 设置无关

from typing_extensions import Annotated

import datetime
from typing import Optional

from sqlalchemy.orm import DeclarativeBase

timestamp = Annotated[
    datetime.datetime,
    mapped_column(nullable=False),
]


class Base(DeclarativeBase):
    pass


class SomeClass(Base):
    # ...

    # pep-484 type will be Optional, but column will be
    # NOT NULL
    created_at: Mapped[Optional[timestamp]]

mapped_column() 构造也与显式传递的 mapped_column() 构造协调,其参数将优先于 Annotated 构造的参数。下面,我们将 ForeignKey 约束添加到我们的整数主键,并且还为 created_at 列使用备用服务器默认值

import datetime

from typing_extensions import Annotated

from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.schema import CreateTable

intpk = Annotated[int, mapped_column(primary_key=True)]
timestamp = Annotated[
    datetime.datetime,
    mapped_column(nullable=False, server_default=func.CURRENT_TIMESTAMP()),
]


class Base(DeclarativeBase):
    pass


class Parent(Base):
    __tablename__ = "parent"

    id: Mapped[intpk]


class SomeClass(Base):
    __tablename__ = "some_table"

    # add ForeignKey to mapped_column(Integer, primary_key=True)
    id: Mapped[intpk] = mapped_column(ForeignKey("parent.id"))

    # change server default from CURRENT_TIMESTAMP to UTC_TIMESTAMP
    created_at: Mapped[timestamp] = mapped_column(server_default=func.UTC_TIMESTAMP())

CREATE TABLE 语句说明了这些每个属性的设置,添加了 FOREIGN KEY 约束,并将 UTC_TIMESTAMP 替换为 CURRENT_TIMESTAMP

>>> from sqlalchemy.schema import CreateTable
>>> print(CreateTable(SomeClass.__table__))
CREATE TABLE some_table ( id INTEGER NOT NULL, created_at DATETIME DEFAULT UTC_TIMESTAMP() NOT NULL, PRIMARY KEY (id), FOREIGN KEY(id) REFERENCES parent (id) )

注意

刚刚描述的 mapped_column() 功能,其中可以使用 PEP 593 Annotated 对象来指示一组完整的列参数,这些对象包含要复制到属性中的“模板” mapped_column() 对象,目前尚未针对其他 ORM 构造(例如 relationship()composite())实现。虽然理论上此功能是可行的,但目前尝试使用 Annotated 来指示 relationship() 和类似内容的进一步参数将在运行时引发 NotImplementedError 异常,但可能会在未来的版本中实现。

在类型映射中使用 Python Enum 或 pep-586 Literal 类型

在版本 2.0.0b4 中新增: - 添加了 Enum 支持

在版本 2.0.1 中新增: - 添加了 Literal 支持

用户定义的 Python 类型,这些类型派生自 Python 内置的 enum.Enum 以及 typing.Literal 类,在 ORM declarative 映射中使用时,会自动链接到 SQLAlchemy Enum 数据类型。下面的示例在 Mapped[] 构造函数中使用自定义的 enum.Enum

import enum

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


class Base(DeclarativeBase):
    pass


class Status(enum.Enum):
    PENDING = "pending"
    RECEIVED = "received"
    COMPLETED = "completed"


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[Status]

在上面的示例中,映射的属性 SomeClass.status 将链接到具有 Enum(Status) 数据类型的 Column。例如,我们可以在 PostgreSQL 数据库的 CREATE TABLE 输出中看到这一点

CREATE TYPE status AS ENUM ('PENDING', 'RECEIVED', 'COMPLETED')

CREATE TABLE some_table (
  id SERIAL NOT NULL,
  status status NOT NULL,
  PRIMARY KEY (id)
)

以类似的方式,也可以使用 typing.Literal,使用由所有字符串组成的 typing.Literal

from typing import Literal

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


class Base(DeclarativeBase):
    pass


Status = Literal["pending", "received", "completed"]


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[Status]

registry.type_annotation_map 中使用的条目将基本 enum.Enum Python 类型以及 typing.Literal 类型链接到 SQLAlchemy Enum SQL 类型,使用一种特殊形式,该形式向 Enum 数据类型指示它应该针对任意枚举类型自动配置自身。默认情况下隐式的此配置将显式指示为

import enum
import typing

import sqlalchemy
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    type_annotation_map = {
        enum.Enum: sqlalchemy.Enum(enum.Enum),
        typing.Literal: sqlalchemy.Enum(enum.Enum),
    }

Declarative 中的解析逻辑能够解析 enum.Enum 的子类以及 typing.Literal 的实例,以匹配 registry.type_annotation_map 字典中的 enum.Enumtyping.Literal 条目。Enum SQL 类型然后知道如何生成具有适当设置(包括默认字符串长度)的自身配置版本。如果传递的 typing.Literal 不仅包含字符串值,则会引发信息丰富的错误。

typing.TypeAliasType 也可以用于创建枚举,方法是将它们分配给字符串的 typing.Literal

from typing import Literal

type Status = Literal["on", "off", "unknown"]

由于这是一个 typing.TypeAliasType,因此它表示唯一的类型对象,因此必须将其放置在 type_annotation_map 中才能成功查找,并键入到 Enum 类型,如下所示

import enum
import sqlalchemy


class Base(DeclarativeBase):
    type_annotation_map = {Status: sqlalchemy.Enum(enum.Enum)}

由于 SQLAlchemy 支持单独映射在结构上等效的不同 typing.TypeAliasType 对象,因此这些对象必须存在于 type_annotation_map 中,以避免歧义。

原生枚举和命名

Enum.native_enum 参数指的是 Enum 数据类型是否应创建所谓的“原生”枚举,在 MySQL/MariaDB 上是 ENUM 数据类型,在 PostgreSQL 上是由 CREATE TYPE 创建的新 TYPE 对象,或者“非原生”枚举,这意味着将使用 VARCHAR 来创建数据类型。对于 MySQL/MariaDB 或 PostgreSQL 以外的后端,在所有情况下都使用 VARCHAR(第三方方言可能有其自身的行为)。

由于 PostgreSQL 的 CREATE TYPE 要求要创建的类型具有显式名称,因此在处理隐式生成的 Enum 而未在映射中指定显式 Enum 数据类型时,存在特殊的后备逻辑

  1. 如果 Enum 链接到 enum.Enum 对象,则 Enum.native_enum 参数默认为 True,枚举的名称将从 enum.Enum 数据类型的名称中获取。PostgreSQL 后端将假定具有此名称的 CREATE TYPE

  2. 如果 Enum 链接到 typing.Literal 对象,则 Enum.native_enum 参数默认为 False;不生成名称,并且假定为 VARCHAR

要将 typing.Literal 与 PostgreSQL CREATE TYPE 类型一起使用,必须使用显式 Enum,可以在类型映射中使用

import enum
import typing

import sqlalchemy
from sqlalchemy.orm import DeclarativeBase

Status = Literal["pending", "received", "completed"]


class Base(DeclarativeBase):
    type_annotation_map = {
        Status: sqlalchemy.Enum("pending", "received", "completed", name="status_enum"),
    }

或者在 mapped_column() 中使用

import enum
import typing

import sqlalchemy
from sqlalchemy.orm import DeclarativeBase

Status = Literal["pending", "received", "completed"]


class Base(DeclarativeBase):
    pass


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[Status] = mapped_column(
        sqlalchemy.Enum("pending", "received", "completed", name="status_enum")
    )
更改默认枚举的配置

为了修改隐式生成的 Enum 数据类型的固定配置,请在 registry.type_annotation_map 中指定新条目,指示其他参数。例如,要无条件地使用“非原生枚举”,可以将所有类型的 Enum.native_enum 参数设置为 False

import enum
import typing
import sqlalchemy
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    type_annotation_map = {
        enum.Enum: sqlalchemy.Enum(enum.Enum, native_enum=False),
        typing.Literal: sqlalchemy.Enum(enum.Enum, native_enum=False),
    }

在版本 2.0.1 中更改: 实现了对在建立 registry.type_annotation_map 时覆盖 Enum 数据类型中的参数(例如 Enum.native_enum)的支持。以前,此功能不起作用。

要为特定的 enum.Enum 子类型使用特定的配置,例如在使用示例 Status 数据类型时将字符串长度设置为 50

import enum
import sqlalchemy
from sqlalchemy.orm import DeclarativeBase


class Status(enum.Enum):
    PENDING = "pending"
    RECEIVED = "received"
    COMPLETED = "completed"


class Base(DeclarativeBase):
    type_annotation_map = {
        Status: sqlalchemy.Enum(Status, length=50, native_enum=False)
    }

默认情况下,自动生成的 Enum 不与 Base 使用的 MetaData 实例关联,因此如果元数据定义了模式,则它不会自动与枚举关联。要自动将枚举与元数据或表(它们所属的)中的模式关联,可以设置 Enum.inherit_schema

from enum import Enum
import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    metadata = sa.MetaData(schema="my_schema")
    type_annotation_map = {Enum: sa.Enum(Enum, inherit_schema=True)}
将特定的 enum.Enumtyping.Literal 链接到其他数据类型

以上示例展示了如何使用一个 Enum,它会自动根据 enum.Enumtyping.Literal 类型对象上存在的参数/属性进行配置。对于应将特定类型的 enum.Enumtyping.Literal 链接到其他类型的使用场景,这些特定类型也可以放置在类型映射中。在下面的示例中,包含非字符串类型的 Literal[] 条目链接到 JSON 数据类型

from typing import Literal

from sqlalchemy import JSON
from sqlalchemy.orm import DeclarativeBase

my_literal = Literal[0, 1, True, False, "true", "false"]


class Base(DeclarativeBase):
    type_annotation_map = {my_literal: JSON}

在上述配置中,my_literal 数据类型将解析为 JSON 实例。其他 Literal 变体将继续解析为 Enum 数据类型。

mapped_column() 中的 Dataclass 功能

mapped_column() 构造与 SQLAlchemy 的“原生 dataclasses”功能集成,详见 声明式 Dataclass 映射。有关 mapped_column() 支持的其他指令的当前背景,请参阅该章节。

访问 Table 和 Metadata

声明式映射类将始终包含一个名为 __table__ 的属性;当使用 __tablename__ 的上述配置完成后,声明式过程会使 Table 通过 __table__ 属性可用

# access the Table
user_table = User.__table__

上面的表最终与 Mapper.local_table 属性对应的表相同,我们可以通过 运行时检查系统 查看。

from sqlalchemy import inspect

user_table = inspect(User).local_table

与声明式 registry 以及基类关联的 MetaData 集合通常是运行 DDL 操作(例如 CREATE)以及与 Alembic 等迁移工具一起使用所必需的。此对象可通过 registry 以及声明式基类的 .metadata 属性获得。下面,对于一个小型脚本,我们可能希望为针对 SQLite 数据库的所有表发出 CREATE 命令

engine = create_engine("sqlite://")

Base.metadata.create_all(engine)

声明式 Table 配置

当使用带有 __tablename__ 声明式类属性的声明式 Table 配置时,要提供给 Table 构造函数的其他参数应使用 __table_args__ 声明式类属性提供。

此属性可以容纳通常发送到 Table 构造函数的位置参数和关键字参数。该属性可以以两种形式之一指定。一种是字典形式:

class MyClass(Base):
    __tablename__ = "sometable"
    __table_args__ = {"mysql_engine": "InnoDB"}

另一种是元组形式,其中每个参数都是位置参数(通常是约束):

class MyClass(Base):
    __tablename__ = "sometable"
    __table_args__ = (
        ForeignKeyConstraint(["id"], ["remote_table.id"]),
        UniqueConstraint("foo"),
    )

可以通过将最后一个参数指定为字典,以上述形式指定关键字参数:

class MyClass(Base):
    __tablename__ = "sometable"
    __table_args__ = (
        ForeignKeyConstraint(["id"], ["remote_table.id"]),
        UniqueConstraint("foo"),
        {"autoload": True},
    )

类还可以使用 declared_attr() 方法装饰器,以动态样式指定 __table_args__ 声明式属性以及 __tablename__ 属性。有关背景信息,请参阅 使用 Mixin 组合映射层次结构

使用声明式 Table 的显式 Schema 名称

指定 Schema 名称 中记录的 Table 的 schema 名称应用于单个 Table,使用 Table.schema 参数。当使用声明式表时,此选项像任何其他选项一样传递到 __table_args__ 字典中

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class MyClass(Base):
    __tablename__ = "sometable"
    __table_args__ = {"schema": "some_schema"}

schema 名称也可以通过使用 MetaData.schema 参数(在 使用 MetaData 指定默认 Schema 名称 中记录)全局应用于所有 Table 对象。MetaData 对象可以单独构造,并通过直接分配给 metadata 属性,与 DeclarativeBase 子类关联

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

metadata_obj = MetaData(schema="some_schema")


class Base(DeclarativeBase):
    metadata = metadata_obj


class MyClass(Base):
    # will use "some_schema" by default
    __tablename__ = "sometable"

另请参阅

指定 Schema 名称 - 在 使用 MetaData 描述数据库 文档中。

为声明式映射列设置加载和持久化选项

mapped_column() 构造接受其他 ORM 特定的参数,这些参数会影响生成的 Column 的映射方式,从而影响其加载和持久化时的行为。常用的选项包括:

  • 延迟列加载 - mapped_column.deferred 布尔值使用 延迟列加载 默认建立 Column。在下面的示例中,User.bio 列默认不会加载,而只会在访问时加载

    class User(Base):
        __tablename__ = "user"
    
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str]
        bio: Mapped[str] = mapped_column(Text, deferred=True)

    另请参阅

    使用列延迟限制加载哪些列 - 延迟列加载的完整描述

  • 活动历史记录 - mapped_column.active_history 确保在属性值更改时,先前的值将被加载并成为检查属性历史记录时 AttributeState.history 集合的一部分。这可能会产生额外的 SQL 语句

    class User(Base):
        __tablename__ = "user"
    
        id: Mapped[int] = mapped_column(primary_key=True)
        important_identifier: Mapped[str] = mapped_column(active_history=True)

有关支持的参数列表,请参阅 mapped_column() 的文档字符串。

另请参阅

为命令式 Table 列应用加载、持久化和映射选项 - 描述了如何将 column_property()deferred() 用于命令式 Table 配置

显式命名声明式映射列

到目前为止的所有示例都展示了连接到 ORM 映射属性的 mapped_column() 构造,其中赋予 mapped_column() 的 Python 属性名称也是我们在 CREATE TABLE 语句和查询中看到的列名。SQL 中表示的列名可以通过传递字符串位置参数 mapped_column.__name 作为第一个位置参数来指示。在下面的示例中,User 类被映射,并为列本身赋予了备用名称

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column("user_id", primary_key=True)
    name: Mapped[str] = mapped_column("user_name")

在上面的示例中,User.id 解析为名为 user_id 的列,而 User.name 解析为名为 user_name 的列。我们可以使用 Python 属性名称编写 select() 语句,并且将看到生成的 SQL 名称

>>> from sqlalchemy import select
>>> print(select(User.id, User.name).where(User.name == "x"))
SELECT "user".user_id, "user".user_name FROM "user" WHERE "user".user_name = :user_name_1

另请参阅

用于映射 Table 列的备用属性名称 - 适用于命令式 Table

向现有声明式映射类追加额外的列

声明式表配置允许在 Table 元数据已生成之后,向现有映射添加新的 Column 对象。

对于使用声明式基类声明的声明式类,底层元类 DeclarativeMeta 包含一个 __setattr__() 方法,该方法将拦截额外的 mapped_column() 或 Core Column 对象,并将它们添加到 Table(使用 Table.append_column())以及现有 Mapper(使用 Mapper.add_property())。

MyClass.some_new_column = mapped_column(String)

使用 core Column

MyClass.some_new_column = Column(String)

支持所有参数,包括备用名称,例如 MyClass.some_new_column = mapped_column("some_name", String)。但是,SQL 类型必须显式传递给 mapped_column()Column 对象,如上述示例中传递 String 类型的情况。 Mapped 注解类型无法参与此操作。

在特定情况下,当使用单表继承时,也可以向映射添加其他 Column 对象,其中映射的子类上存在额外的列,而这些子类没有自己的 Table。这在 单表继承 章节中进行了说明。

另请参阅

在声明后向映射类添加关系 - relationship() 的类似示例

注意

仅当使用“声明式基类”(意味着 DeclarativeBase 的用户定义子类或由 declarative_base()registry.generate_base() 返回的动态生成的类)时,将映射的属性分配给已映射的类才能正常工作。这个“基类”包含一个 Python 元类,该元类实现了一个特殊的 __setattr__() 方法来拦截这些操作。

如果使用装饰器(如 registry.mapped())或命令式函数(如 registry.map_imperatively())映射类,则运行时将类映射属性分配给映射类将不起作用

声明式与命令式 Table(又名混合声明式)

声明式映射也可以使用预先存在的 Table 对象,或者其他单独构造的 Table 或其他任意 FromClause 构造(例如 JoinSubquery)。

这被称为“混合声明式”映射,因为该类使用声明式风格映射所有涉及映射器配置的内容,但是映射的 Table 对象是单独生成的,并直接传递给声明式过程

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


# construct a Table directly.  The Base.metadata collection is
# usually a good choice for MetaData but any MetaData
# collection may be used.

user_table = Table(
    "user",
    Base.metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("fullname", String),
    Column("nickname", String),
)


# construct the User class using this table.
class User(Base):
    __table__ = user_table

上面,Table 对象是使用 使用 MetaData 描述数据库 中描述的方法构造的。然后,它可以直接应用于声明式映射的类。__tablename____table_args__ 声明式类属性在这种形式下不使用。以上配置通常更易于作为内联定义阅读:

class User(Base):
    __table__ = Table(
        "user",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String),
        Column("fullname", String),
        Column("nickname", String),
    )

上述样式的自然效果是 __table__ 属性本身在类定义块内定义。因此,它可以立即在后续属性中引用,例如下面的示例说明了在多态映射器配置中引用 type 列的情况

class Person(Base):
    __table__ = Table(
        "person",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50)),
        Column("type", String(50)),
    )

    __mapper_args__ = {
        "polymorphic_on": __table__.c.type,
        "polymorphic_identity": "person",
    }

当要映射非 Table 构造(例如 JoinSubquery 对象)时,也使用“命令式表”形式。下面是一个示例:

from sqlalchemy import func, select

subq = (
    select(
        func.count(orders.c.id).label("order_count"),
        func.max(orders.c.price).label("highest_order"),
        orders.c.customer_id,
    )
    .group_by(orders.c.customer_id)
    .subquery()
)

customer_select = (
    select(customers, subq)
    .join_from(customers, subq, customers.c.id == subq.c.customer_id)
    .subquery()
)


class Customer(Base):
    __table__ = customer_select

有关映射到非 Table 构造的背景信息,请参阅 将类映射到多个表将类映射到任意子查询 章节。

当类本身使用属性声明的替代形式(例如 Python dataclasses)时,“命令式表”形式特别有用。有关详细信息,请参阅 将 ORM 映射应用于现有 dataclass(旧版 dataclass 用法) 章节。

用于映射 Table 列的备用属性名称

显式命名声明式映射列 章节说明了如何使用 mapped_column() 为生成的 Column 对象提供一个特定的名称,该名称与映射它的属性名称分开。

当使用命令式 Table 配置时,我们已经有了 Column 对象。要将这些对象映射到备用名称,我们可以直接将 Column 分配给所需的属性:

user_table = Table(
    "user",
    Base.metadata,
    Column("user_id", Integer, primary_key=True),
    Column("user_name", String),
)


class User(Base):
    __table__ = user_table

    id = user_table.c.user_id
    name = user_table.c.user_name

上面的 User 映射将通过 User.idUser.name 属性引用 "user_id""user_name" 列,与 显式命名声明式映射列 中演示的方式相同。

上述映射的一个注意事项是,当使用 PEP 484 类型工具时,直接内联链接到 Column 将无法正确键入。解决此问题的一种策略是在 column_property() 函数中应用 Column 对象;虽然 Mapper 已经为其内部使用自动生成此属性对象,但是通过在类声明中命名它,类型工具将能够将属性与 Mapped 注解匹配

from sqlalchemy.orm import column_property
from sqlalchemy.orm import Mapped


class User(Base):
    __table__ = user_table

    id: Mapped[int] = column_property(user_table.c.user_id)
    name: Mapped[str] = column_property(user_table.c.user_name)

另请参阅

显式命名声明式映射列 - 适用于声明式 Table

为命令式 Table 列应用加载、持久化和映射选项

为声明式映射列设置加载和持久化选项 章节回顾了在使用带有声明式 Table 配置的 mapped_column() 构造时如何设置加载和持久化选项。当使用命令式 Table 配置时,我们已经有了要映射的现有 Column 对象。为了将这些 Column 对象与 ORM 映射特有的其他参数一起映射,我们可以使用 column_property()deferred() 构造,以便将其他参数与列关联。选项包括:

  • 延迟列加载 - deferred() 函数是调用 column_property() 并将 column_property.deferred 参数设置为 True 的简写形式;此构造使用 延迟列加载 默认建立 Column。在下面的示例中,User.bio 列默认不会加载,而只会在访问时加载

    from sqlalchemy.orm import deferred
    
    user_table = Table(
        "user",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String),
        Column("bio", Text),
    )
    
    
    class User(Base):
        __table__ = user_table
    
        bio = deferred(user_table.c.bio)

另请参阅

使用列延迟限制加载哪些列 - 延迟列加载的完整描述

  • 活动历史记录 - column_property.active_history 确保在属性值更改时,先前的值将被加载并成为检查属性历史记录时 AttributeState.history 集合的一部分。这可能会产生额外的 SQL 语句

    from sqlalchemy.orm import deferred
    
    user_table = Table(
        "user",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("important_identifier", String),
    )
    
    
    class User(Base):
        __table__ = user_table
    
        important_identifier = column_property(
            user_table.c.important_identifier, active_history=True
        )

另请参阅

column_property() 构造对于将类映射到备用 FROM 子句(如 joins 和 selects)的情况也很重要。有关这些情况的更多背景信息,请参阅:

对于带有 mapped_column() 的声明式 Table 配置,大多数选项都可直接使用;有关示例,请参阅 为声明式映射列设置加载和持久化选项 章节。

使用反射表进行声明式映射

这里有几种可用的模式,可以针对从数据库内省出的一系列 Table 对象生成映射类,使用的内省过程在 反射数据库对象 中描述。

将类映射到从数据库反射出的表的一个简单方法是使用声明式混合映射,并将 Table.autoload_with 参数传递给 Table 的构造函数

from sqlalchemy import create_engine
from sqlalchemy import Table
from sqlalchemy.orm import DeclarativeBase

engine = create_engine("postgresql+psycopg2://user:pass@hostname/my_existing_database")


class Base(DeclarativeBase):
    pass


class MyClass(Base):
    __table__ = Table(
        "mytable",
        Base.metadata,
        autoload_with=engine,
    )

上述模式的一种变体可以扩展到许多表,即使用 MetaData.reflect() 方法一次性反射出一整套 Table 对象,然后从 MetaData 中引用它们

from sqlalchemy import create_engine
from sqlalchemy import Table
from sqlalchemy.orm import DeclarativeBase

engine = create_engine("postgresql+psycopg2://user:pass@hostname/my_existing_database")


class Base(DeclarativeBase):
    pass


Base.metadata.reflect(engine)


class MyClass(Base):
    __table__ = Base.metadata.tables["mytable"]

使用 __table__ 方式的一个注意事项是,映射类必须在表被反射之后才能声明,这要求在声明应用程序类时数据库连接源必须存在;通常情况下,类是在应用程序模块被导入时声明的,但数据库连接直到应用程序开始运行代码才能使用,以便它可以消费配置信息并创建引擎。目前有两种方法可以解决这个问题,将在接下来的两节中描述。

使用 DeferredReflection

为了适应在之后进行表元数据反射的情况下声明映射类的用例,一个名为 DeferredReflection 混入的简单扩展可用,它改变了声明式映射过程,使其延迟到调用特殊的类级别 DeferredReflection.prepare() 方法时才执行,该方法将对目标数据库执行反射过程,并将结果与声明式表映射过程集成,即使用 __tablename__ 属性的类

from sqlalchemy.ext.declarative import DeferredReflection
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class Reflected(DeferredReflection):
    __abstract__ = True


class Foo(Reflected, Base):
    __tablename__ = "foo"
    bars = relationship("Bar")


class Bar(Reflected, Base):
    __tablename__ = "bar"

    foo_id = mapped_column(Integer, ForeignKey("foo.id"))

在上面,我们创建了一个混入类 Reflected,它将作为我们的声明式层次结构中类的基类,这些类应该在调用 Reflected.prepare 方法时被映射。只有在我们给定一个 Engine 并这样做之后,上面的映射才是完整的

engine = create_engine("postgresql+psycopg2://user:pass@hostname/my_existing_database")
Reflected.prepare(engine)

Reflected 类的目的是定义应该以反射方式映射的类的范围。该插件将在调用 .prepare() 的目标类的子类树中搜索,并反射所有由声明的类命名的表;目标数据库中不属于映射且与目标表没有外键约束关系的表将不会被反射。

使用 Automap

对于针对现有数据库进行映射,并使用表反射的更自动化解决方案是使用 Automap 扩展。此扩展将从数据库模式生成完整的映射类,包括基于观察到的外键约束的类之间的关系。虽然它包括用于自定义的钩子,例如允许自定义类命名和关系命名方案的钩子,但 automap 的目标是实现快速的零配置工作方式。如果应用程序希望拥有一个完全显式的模型,该模型利用表反射,那么 DeferredReflection 类可能是更可取的,因为它自动化程度较低。

另请参阅

Automap

自动化来自反射表的列命名方案

当使用任何先前的反射技术时,我们可以选择更改列映射的命名方案。Column 对象包含一个参数 Column.key,它是一个字符串名称,用于确定此 ColumnTable.c 集合中存在的名称,独立于列的 SQL 名称。如果未通过其他方式(例如在 映射表列的备用属性名称 中说明的方式)提供此键,则 Mapper 也将此键用作属性名称,在该名称下映射 Column

当使用表反射时,我们可以拦截用于 Column 的参数,因为它们是使用 DDLEvents.column_reflect() 事件接收的,并应用我们需要做的任何更改,包括 .key 属性以及数据类型等。

事件钩子最容易与正在使用的 MetaData 对象关联,如下所示

from sqlalchemy import event
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


@event.listens_for(Base.metadata, "column_reflect")
def column_reflect(inspector, table, column_info):
    # set column.key = "attr_<lower_case_name>"
    column_info["key"] = "attr_%s" % column_info["name"].lower()

通过上面的事件,Column 对象的反射将被我们的事件拦截,该事件添加了一个新的 “.key” 元素,例如在下面的映射中

class MyClass(Base):
    __table__ = Table("some_table", Base.metadata, autoload_with=some_engine)

该方法也适用于 DeferredReflection 基类以及 Automap 扩展。对于 automap,请特别参阅 拦截列定义 部分以了解背景信息。

映射到显式主键列集

Mapper 构造为了成功映射表,始终要求至少有一列被标识为该可选对象的“主键”。这是为了当加载或持久化 ORM 对象时,它可以放置在 标识映射 中,并具有适当的 标识键

在那些要映射的反射表不包含主键约束的情况下,以及在 针对任意可选对象进行映射 的一般情况下,可能不存在主键列,因此提供了 Mapper.primary_key 参数,以便可以将任何列集配置为表的“主键”,就 ORM 映射而言。

给定以下针对现有 Table 对象(表可能在反射场景中没有声明任何主键)的命令式表映射示例,我们可以像以下示例中那样映射这样的表

from sqlalchemy import Column
from sqlalchemy import MetaData
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import DeclarativeBase


metadata = MetaData()
group_users = Table(
    "group_users",
    metadata,
    Column("user_id", String(40), nullable=False),
    Column("group_id", String(40), nullable=False),
    UniqueConstraint("user_id", "group_id"),
)


class Base(DeclarativeBase):
    pass


class GroupUsers(Base):
    __table__ = group_users
    __mapper_args__ = {"primary_key": [group_users.c.user_id, group_users.c.group_id]}

在上面,group_users 表是一个某种类型的关联表,具有字符串列 user_idgroup_id,但未设置主键;相反,只有一个 UniqueConstraint 建立这两列代表唯一键。Mapper 不会自动检查唯一约束以查找主键;相反,我们使用 Mapper.primary_key 参数,传递 [group_users.c.user_id, group_users.c.group_id] 集合,指示应使用这两列来构造 GroupUsers 类的实例的标识键。

映射表列的子集

有时,表反射可能会提供一个 Table,其中包含许多对我们来说不重要且可以安全忽略的列。对于这样一个有很多列的表,这些列不需要在应用程序中引用,Mapper.include_propertiesMapper.exclude_properties 参数可以指示要映射的列的子集,其中目标 Table 中的其他列将不会以任何方式被 ORM 考虑。例子

class User(Base):
    __table__ = user_table
    __mapper_args__ = {"include_properties": ["user_id", "user_name"]}

在上面的示例中,User 类将映射到 user_table 表,仅包括 user_iduser_name 列 - 其余的列不被引用。

类似地

class Address(Base):
    __table__ = address_table
    __mapper_args__ = {"exclude_properties": ["street", "city", "state", "zip"]}

Address 类映射到 address_table 表,包括除 streetcitystatezip 之外的所有列。

如两个示例所示,列可以通过字符串名称或直接引用 Column 对象来引用。直接引用对象对于显式性以及在映射到可能具有重复名称的多表构造时解决歧义可能很有用

class User(Base):
    __table__ = user_table
    __mapper_args__ = {
        "include_properties": [user_table.c.user_id, user_table.c.user_name]
    }

当列未包含在映射中时,这些列将不会在执行 select() 或旧式 Query 对象时发出的任何 SELECT 语句中引用,并且映射类上也不会有任何表示该列的映射属性;分配该名称的属性除了正常的 Python 属性分配之外,不会有任何效果。

但是,重要的是要注意,对于那些包含模式级别列默认值的 Column 对象,模式级别列默认值仍然有效,即使它们可能从 ORM 映射中排除。

“模式级别列默认值”是指在 列 INSERT/UPDATE 默认值 中描述的默认值,包括由 Column.defaultColumn.onupdateColumn.server_defaultColumn.server_onupdate 参数配置的默认值。这些构造继续具有正常效果,因为在 Column.defaultColumn.onupdate 的情况下,Column 对象仍然存在于底层的 Table 上,从而允许默认函数在 ORM 发出 INSERT 或 UPDATE 时生效,并且在 Column.server_defaultColumn.server_onupdate 的情况下,关系数据库本身将这些默认值作为服务器端行为发出。