SqlAlchemy 2.0 中文文档(六)(4)

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS PostgreSQL,集群系列 2核4GB
简介: SqlAlchemy 2.0 中文文档(六)

SqlAlchemy 2.0 中文文档(六)(3)https://developer.aliyun.com/article/1560702


使用 declared_attr() 生成特定于表的继承列

与在与 declared_attr 一起使用时如何处理 __tablename__ 和其他特殊名称不同,当我们混入列和属性(例如关系、列属性等)时,该函数仅在层次结构中的基类调用,除非结合使用 declared_attr 指令和 declared_attr.cascading 子指令。在下面的示例中,只有 Person 类将收到名为 id 的列;对于未给出主键的 Engineer,映射将失败:

class HasId:
    id: Mapped[int] = mapped_column(primary_key=True)
class Person(HasId, Base):
    __tablename__ = "person"
    discriminator: Mapped[str]
    __mapper_args__ = {"polymorphic_on": "discriminator"}
# this mapping will fail, as there's no primary key
class Engineer(Person):
    __tablename__ = "engineer"
    primary_language: Mapped[str]
    __mapper_args__ = {"polymorphic_identity": "engineer"}

在连接表继承中,通常情况下我们希望每个子类具有不同命名的列。然而在这种情况下,我们可能希望每个表上都有一个 id 列,并且通过外键相互引用。我们可以通过使用 declared_attr.cascading 修饰符作为 mixin 来实现这一点,该修饰符表示应该对层次结构中的每个类调用该函数,几乎(见下面的警告)与对 __tablename__ 调用方式相同:

class HasIdMixin:
    @declared_attr.cascading
    def id(cls) -> Mapped[int]:
        if has_inherited_table(cls):
            return mapped_column(ForeignKey("person.id"), primary_key=True)
        else:
            return mapped_column(Integer, primary_key=True)
class Person(HasIdMixin, Base):
    __tablename__ = "person"
    discriminator: Mapped[str]
    __mapper_args__ = {"polymorphic_on": "discriminator"}
class Engineer(Person):
    __tablename__ = "engineer"
    primary_language: Mapped[str]
    __mapper_args__ = {"polymorphic_identity": "engineer"}

警告

目前,declared_attr.cascading 功能不允许子类使用不同的函数或值覆盖属性。这是在如何解析 @declared_attr 的机制中的当前限制,并且如果检测到此条件,则会发出警告。此限制仅适用于 ORM 映射的列、关系和其他属性的MapperProperty风格。它适用于诸如 __tablename____mapper_args__ 等的声明性指令,这些指令在内部以与declared_attr.cascading不同的方式解析。

从多个混合类组合表/映射器参数

当使用声明性混合类指定 __table_args____mapper_args__ 时,您可能希望将一些参数从多个混合类中与您希望在类本身上定义的参数结合起来。可以在这里使用 declared_attr 装饰器来创建用户定义的排序例程,该例程从多个集合中提取:

from sqlalchemy.orm import declarative_mixin, declared_attr
class MySQLSettings:
    __table_args__ = {"mysql_engine": "InnoDB"}
class MyOtherMixin:
    __table_args__ = {"info": "foo"}
class MyModel(MySQLSettings, MyOtherMixin, Base):
    __tablename__ = "my_model"
    @declared_attr.directive
    def __table_args__(cls):
        args = dict()
        args.update(MySQLSettings.__table_args__)
        args.update(MyOtherMixin.__table_args__)
        return args
    id = mapped_column(Integer, primary_key=True)

在混合类上使用命名约定创建索引和约束

对于具有命名约束的使用,如 IndexUniqueConstraintCheckConstraint,其中每个对象应该是唯一的,针对从混合类派生的特定表,需要为每个实际映射的类创建每个对象的单独实例。

作为一个简单的示例,要定义一个命名的、可能是多列的 Index,该索引适用于从混合类派生的所有表,可以使用 Index 的“内联”形式,并将其建立为 __table_args__ 的一部分,使用 declared_attr 来建立 __table_args__() 作为一个类方法,该方法将被每个子类调用:

class MyMixin:
    a = mapped_column(Integer)
    b = mapped_column(Integer)
    @declared_attr.directive
    def __table_args__(cls):
        return (Index(f"test_idx_{cls.__tablename__}", "a", "b"),)
class MyModelA(MyMixin, Base):
    __tablename__ = "table_a"
    id = mapped_column(Integer, primary_key=True)
class MyModelB(MyMixin, Base):
    __tablename__ = "table_b"
    id = mapped_column(Integer, primary_key=True)

上面的示例将生成两个表 "table_a""table_b",其中包含索引 "test_idx_table_a""test_idx_table_b"

通常,在现代 SQLAlchemy 中,我们会使用一种命名约定,如在配置约束命名约定中记录的那样。虽然命名约定会在创建新的Constraint对象时自动进行,因为此约定是在基于特定Constraint的父Table的对象构造时间应用的,因此需要为每个继承子类创建一个不同的Constraint对象,并再次使用declared_attr__table_args__(),下面通过使用抽象映射基类进行说明:

from uuid import UUID
from sqlalchemy import CheckConstraint
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
constraint_naming_conventions = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}
class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=constraint_naming_conventions)
class MyAbstractBase(Base):
    __abstract__ = True
    @declared_attr.directive
    def __table_args__(cls):
        return (
            UniqueConstraint("uuid"),
            CheckConstraint("x > 0 OR y < 100", name="xy_chk"),
        )
    id: Mapped[int] = mapped_column(primary_key=True)
    uuid: Mapped[UUID]
    x: Mapped[int]
    y: Mapped[int]
class ModelAlpha(MyAbstractBase):
    __tablename__ = "alpha"
class ModelBeta(MyAbstractBase):
    __tablename__ = "beta"

上述映射将生成包含所有约束的表特定名称的 DDL,包括主键、CHECK 约束、唯一约束:

CREATE  TABLE  alpha  (
  id  INTEGER  NOT  NULL,
  uuid  CHAR(32)  NOT  NULL,
  x  INTEGER  NOT  NULL,
  y  INTEGER  NOT  NULL,
  CONSTRAINT  pk_alpha  PRIMARY  KEY  (id),
  CONSTRAINT  uq_alpha_uuid  UNIQUE  (uuid),
  CONSTRAINT  ck_alpha_xy_chk  CHECK  (x  >  0  OR  y  <  100)
)
CREATE  TABLE  beta  (
  id  INTEGER  NOT  NULL,
  uuid  CHAR(32)  NOT  NULL,
  x  INTEGER  NOT  NULL,
  y  INTEGER  NOT  NULL,
  CONSTRAINT  pk_beta  PRIMARY  KEY  (id),
  CONSTRAINT  uq_beta_uuid  UNIQUE  (uuid),
  CONSTRAINT  ck_beta_xy_chk  CHECK  (x  >  0  OR  y  <  100)
)

增强基类

除了使用纯混合外,本节中的大多数技术也可以直接应用于基类,用于适用于从特定基类派生的所有类的模式。下面的示例说明了如何在Base类方面应用上一节的一些示例:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
  """define a series of common elements that may be applied to mapped
 classes using this class as a base class."""
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower()
    __table_args__ = {"mysql_engine": "InnoDB"}
    __mapper_args__ = {"eager_defaults": True}
    id: Mapped[int] = mapped_column(primary_key=True)
class HasLogRecord:
  """mark classes that have a many-to-one relationship to the
 ``LogRecord`` class."""
    log_record_id: Mapped[int] = mapped_column(ForeignKey("logrecord.id"))
    @declared_attr
    def log_record(self) -> Mapped["LogRecord"]:
        return relationship("LogRecord")
class LogRecord(Base):
    log_info: Mapped[str]
class MyModel(HasLogRecord, Base):
    name: Mapped[str]

在上述示例中,MyModelLogRecord,在派生自Base时,它们的表名都将根据其类名派生,一个名为id的主键列,以及由Base.__table_args__Base.__mapper_args__定义的上述表和映射器参数。

在使用旧版declarative_base()registry.generate_base()时,可以使用declarative_base.cls参数来生成等效效果,如下所示的未注释示例:

# legacy declarative_base() use
from sqlalchemy import Integer, String
from sqlalchemy import ForeignKey
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
class Base:
  """define a series of common elements that may be applied to mapped
 classes using this class as a base class."""
    @declared_attr.directive
    def __tablename__(cls):
        return cls.__name__.lower()
    __table_args__ = {"mysql_engine": "InnoDB"}
    __mapper_args__ = {"eager_defaults": True}
    id = mapped_column(Integer, primary_key=True)
Base = declarative_base(cls=Base)
class HasLogRecord:
  """mark classes that have a many-to-one relationship to the
 ``LogRecord`` class."""
    log_record_id = mapped_column(ForeignKey("logrecord.id"))
    @declared_attr
    def log_record(self):
        return relationship("LogRecord")
class LogRecord(Base):
    log_info = mapped_column(String)
class MyModel(HasLogRecord, Base):
    name = mapped_column(String)

混合列

如果使用声明式表的配置风格(而不是命令式表配置),则可以在混合类中指示列,以便在声明式过程生成的 Table 的一部分。可以在声明式混合类中内联声明三种构造:mapped_column()MappedColumn

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime]
class MyModel(TimestampMixin, Base):
    __tablename__ = "test"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

在上述情况下,所有包括 TimestampMixin 在其类基类中的声明式类将自动包含一个名为 created_at 的列,该列对所有行插入应用时间戳,以及一个名为 updated_at 的列,该列不包含默认值以示例为目的(如果有的话,我们将使用 Column.onupdate 参数,该参数被 mapped_column() 接受)。这些列构造始终从源混合类或基类复制,因此可以将相同的混合类/基类应用于任意数量的目标类,每个目标类都将有自己的列构造。

所有声明式列形式都受到混合类的支持,包括:

  • 带注释的属性 - 无论是否存在 mapped_column()
class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime]
  • mapped_column - 无论是否存在 Mapped
class TimestampMixin:
    created_at = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column()
  • Column - 传统的声明式形式:
class TimestampMixin:
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime)

在上述每种形式中,声明式通过创建构造的副本来处理混合类上的基于列的属性,然后将其应用于目标类。

版本 2.0 中的变化:声明式 API 现在可以接受 Column 对象以及任何形式的 mapped_column() 构造,当使用混合类时无需使用 declared_attr()。已经移除了以前的限制,这些限制阻止直接在混合类中使用具有 ForeignKey 元素的列。

混入关系

通过relationship()创建的关系通过declared_attr方法提供的声明式混合类,排除了在复制关系及其可能与列绑定的内容时可能出现的任何歧义。下面是一个示例,将外键列和关系组合在一起,以便两个类FooBar都可以配置为通过多对一引用一个共同的目标类:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
    pass
class RefTargetMixin:
    target_id: Mapped[int] = mapped_column(ForeignKey("target.id"))
    @declared_attr
    def target(cls) -> Mapped["Target"]:
        return relationship("Target")
class Foo(RefTargetMixin, Base):
    __tablename__ = "foo"
    id: Mapped[int] = mapped_column(primary_key=True)
class Bar(RefTargetMixin, Base):
    __tablename__ = "bar"
    id: Mapped[int] = mapped_column(primary_key=True)
class Target(Base):
    __tablename__ = "target"
    id: Mapped[int] = mapped_column(primary_key=True)

使用上述映射,每个FooBar都包含一个到Target的关系,通过.target属性访问:

>>> from sqlalchemy import select
>>> print(select(Foo).join(Foo.target))
SELECT  foo.id,  foo.target_id
FROM  foo  JOIN  target  ON  target.id  =  foo.target_id
>>> print(select(Bar).join(Bar.target))
SELECT  bar.id,  bar.target_id
FROM  bar  JOIN  target  ON  target.id  =  bar.target_id 

类似relationship.primaryjoin的特殊参数也可以在混入的 classmethod 中使用,这些参数通常需要引用正在映射的类。对于需要引用本地映射列的方案,在普通情况下,这些列通过 Declarative 作为映射类的属性提供,该类作为参数cls传递给修饰的 classmethod。使用此功能,我们可以例如使用显式的 primaryjoin 重写RefTargetMixin.target方法,该方法引用了Targetcls上的待定映射列:

class Target(Base):
    __tablename__ = "target"
    id: Mapped[int] = mapped_column(primary_key=True)
class RefTargetMixin:
    target_id: Mapped[int] = mapped_column(ForeignKey("target.id"))
    @declared_attr
    def target(cls) -> Mapped["Target"]:
        # illustrates explicit 'primaryjoin' argument
        return relationship("Target", primaryjoin=Target.id == cls.target_id)

混合使用column_property()和其他MapperProperty

relationship()类似,其他MapperProperty子类,如column_property()在被混合使用时也需要生成类局部副本,因此在被declared_attr修饰的函数内声明。在该函数内,使用mapped_column()MappedColumn声明的其他普通映射列将从cls参数中提取,以便它们可以被用来组合新的属性,如下例所示,将两个列相加:

from sqlalchemy.orm import column_property
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(DeclarativeBase):
    pass
class SomethingMixin:
    x: Mapped[int]
    y: Mapped[int]
    @declared_attr
    def x_plus_y(cls) -> Mapped[int]:
        return column_property(cls.x + cls.y)
class Something(SomethingMixin, Base):
    __tablename__ = "something"
    id: Mapped[int] = mapped_column(primary_key=True)

在上面的例子中,我们可以在生成完整表达式的语句中使用Something.x_plus_y

>>> from sqlalchemy import select
>>> print(select(Something.x_plus_y))
SELECT  something.x  +  something.y  AS  anon_1
FROM  something 

提示

declared_attr装饰器使装饰的可调用对象表现得完全像一个类方法。然而,像Pylance这样的类型工具可能无法识别这一点,这有时会导致它在函数体内部抱怨无法访问cls变量。当出现此问题时,可以直接将@classmethod装饰器与declared_attr结合使用来解决:

class SomethingMixin:
    x: Mapped[int]
    y: Mapped[int]
    @declared_attr
    @classmethod
    def x_plus_y(cls) -> Mapped[int]:
        return column_property(cls.x + cls.y)

新版本 2.0 中:- declared_attr可以容纳使用@classmethod装饰的函数,以帮助需要的PEP 484集成。


SqlAlchemy 2.0 中文文档(六)(5)https://developer.aliyun.com/article/1560712

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
相关文章
|
5月前
|
SQL 测试技术 API
SqlAlchemy 2.0 中文文档(一)(1)
SqlAlchemy 2.0 中文文档(一)
177 1
SqlAlchemy 2.0 中文文档(一)(1)
|
5月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档(一)(3)
SqlAlchemy 2.0 中文文档(一)
76 1
|
5月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档(一)(4)
SqlAlchemy 2.0 中文文档(一)
70 1
|
5月前
|
SQL API 数据库
SqlAlchemy 2.0 中文文档(一)(5)
SqlAlchemy 2.0 中文文档(一)
116 1
|
5月前
|
SQL 数据库 Python
SqlAlchemy 2.0 中文文档(十)(3)
SqlAlchemy 2.0 中文文档(十)
36 1
|
5月前
|
SQL 存储 API
SqlAlchemy 2.0 中文文档(十)(5)
SqlAlchemy 2.0 中文文档(十)
33 1
|
5月前
|
SQL 测试技术 Python
SqlAlchemy 2.0 中文文档(二)(1)
SqlAlchemy 2.0 中文文档(二)
55 2
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(五)(1)
SqlAlchemy 2.0 中文文档(五)
46 0
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(十)(2)
SqlAlchemy 2.0 中文文档(十)
26 0
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档(五)(3)
SqlAlchemy 2.0 中文文档(五)
48 0