SqlAlchemy 2.0 中文文档（十）（2）

SqlAlchemy 2.0 中文文档（十）（1）https://developer.aliyun.com/article/1562915

编程或条件版本计数器

当version_id_generator设置为 False 时，我们还可以以与分配任何其他映射属性相同的方式，在对象上编程（和有条件地）设置版本标识符。例如，如果我们使用了 UUID 示例，但将version_id_generator设置为False，我们可以随意设置版本标识符：

import uuid
class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    version_uuid = mapped_column(String(32), nullable=False)
    name = mapped_column(String(50), nullable=False)
    __mapper_args__ = {"version_id_col": version_uuid, "version_id_generator": False}
u1 = User(name="u1", version_uuid=uuid.uuid4())
session.add(u1)
session.commit()
u1.name = "u2"
u1.version_uuid = uuid.uuid4()
session.commit()

我们也可以在不增加版本计数器的情况下更新我们的User对象；计数器的值将保持不变，并且 UPDATE 语句仍将针对先前的值进行检查。对于仅某些类别的 UPDATE 对并发问题敏感的方案，这可能很有用：

# will leave version_uuid unchanged
u1.name = "u3"
session.commit()

简单版本计数

跟踪版本的最直接方法是向映射表添加一个整数列，然后将其设置为映射选项中的version_id_col：

class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    version_id = mapped_column(Integer, nullable=False)
    name = mapped_column(String(50), nullable=False)
    __mapper_args__ = {"version_id_col": version_id}

注意

强烈建议将version_id列设置为 NOT NULL。版本控制功能不支持版本控制列中的 NULL 值。

上面，User映射使用列version_id跟踪整数版本。当首次刷新User类型的对象时，version_id列的值将为“1”。然后，稍后对表的 UPDATE 将始终以类似以下方式发出：

UPDATE  user  SET  version_id=:version_id,  name=:name
WHERE  user.id  =  :user_id  AND  user.version_id  =  :user_version_id
-- {"name": "new name", "version_id": 2, "user_id": 1, "user_version_id": 1}

上述 UPDATE 语句正在更新不仅与 user.id = 1 匹配的行，而且还要求 user.version_id = 1，其中“1”是我们已知的此对象上一次使用的最后版本标识符。如果其他地方的事务独立修改了行，则此版本  ID 将不再匹配，UPDATE 语句将报告没有匹配的行；这是 SQLAlchemy 测试的条件，确保我们的 UPDATE（或  DELETE）语句仅匹配了一行。如果没有匹配的行，则表示我们的数据版本已过期，并且会引发 StaleDataError。

自定义版本计数器 / 类型

其他类型的值或计数器可以用于版本控制。常见的类型包括日期和 GUID。当使用替代类型或计数器方案时，SQLAlchemy 提供了一个钩子来使用 version_id_generator 参数，该参数接受版本生成可调用对象。此可调用对象将传递当前已知版本的值，并且预计返回后续版本。

例如，如果我们想要使用随机生成的 GUID 跟踪我们的 User 类的版本控制，我们可以这样做（请注意，一些后端支持原生的 GUID 类型，但我们在这里使用简单的字符串进行演示）：

import uuid
class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    version_uuid = mapped_column(String(32), nullable=False)
    name = mapped_column(String(50), nullable=False)
    __mapper_args__ = {
        "version_id_col": version_uuid,
        "version_id_generator": lambda version: uuid.uuid4().hex,
    }

持久性引擎每次将 User 对象受到 INSERT 或 UPDATE 影响时都会调用 uuid.uuid4()。在这种情况下，我们的版本生成函数可以忽略 version 的传入值，因为 uuid4() 函数生成的标识符没有任何先决条件值。如果我们使用的是顺序版本控制方案，例如数字或特殊字符系统，则可以利用给定的 version 来帮助确定后续值。

另请参阅

不特定后端的 GUID 类型

服务器端版本计数器

version_id_generator 也可以配置为依赖于数据库生成的值。在这种情况下，数据库需要在将行受到  INSERT 时以及 UPDATE 时生成新标识符的某种手段。对于 UPDATE  情况，通常需要一个更新触发器，除非所涉及的数据库支持其他本地版本标识符。特别是，PostgreSQL 数据库支持一个名为 xmin 的系统列，提供 UPDATE 版本控制。我们可以利用 PostgreSQL 的 xmin 列来为我们的 User 类进行版本控制，如下所示：

from sqlalchemy import FetchedValue
class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(50), nullable=False)
    xmin = mapped_column("xmin", String, system=True, server_default=FetchedValue())
    __mapper_args__ = {"version_id_col": xmin, "version_id_generator": False}

使用上述映射，ORM 将依赖于 xmin 列来自动提供版本 ID 计数器的新值。

当 ORM 发出 INSERT 或 UPDATE 时，通常不会主动获取数据库生成的值的值，而是将这些列保留为“过期”，并在下次访问它们时获取，除非设置了 eager_defaults Mapper 标志。然而，当使用服务器端版本列时，ORM 需要主动获取新生成的值。这是为了在任何并发事务可能再次更新它之前设置版本计数器。最好在  INSERT 或 UPDATE 语句中同时进行这个获取，使用 RETURNING，否则，如果之后发出一个 SELECT  语句，那么版本计数器在它被获取之前可能会发生竞争条件。

当目标数据库支持 RETURNING 时，我们的 User 类的 INSERT 语句将如下所示：

INSERT  INTO  "user"  (name)  VALUES  (%(name)s)  RETURNING  "user".id,  "user".xmin
-- {'name': 'ed'}

在上述情况下，ORM 可以在一个语句中获取任何新生成的主键值以及服务器生成的版本标识符。当后端不支持 RETURNING 时，必须为每个 INSERT 和 UPDATE 发出额外的 SELECT，这非常低效，还会引入遗漏版本计数器的可能性：

INSERT  INTO  "user"  (name)  VALUES  (%(name)s)
-- {'name': 'ed'}
SELECT  "user".version_id  AS  user_version_id  FROM  "user"  where
"user".id  =  :param_1
-- {"param_1": 1}

仅在绝对必要时，并且仅在支持返回的后端上，强烈建议仅使用服务器端版本计数器，目前支持的后端有 PostgreSQL、Oracle、MariaDB 10.5、SQLite 3.35 和 SQL Server。

编程或有条件的版本计数器

当 version_id_generator 设置为 False 时，我们也可以以编程方式（并有条件地）像分配任何其他映射属性一样，在对象上设置版本标识符。例如，如果我们使用了 UUID 示例，但将 version_id_generator 设置为 False，我们可以根据自己的需要设置版本标识符：

import uuid
class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    version_uuid = mapped_column(String(32), nullable=False)
    name = mapped_column(String(50), nullable=False)
    __mapper_args__ = {"version_id_col": version_uuid, "version_id_generator": False}
u1 = User(name="u1", version_uuid=uuid.uuid4())
session.add(u1)
session.commit()
u1.name = "u2"
u1.version_uuid = uuid.uuid4()
session.commit()

我们还可以在不递增版本计数器的情况下更新我们的 User 对象；计数器的值将保持不变，并且 UPDATE 语句仍将根据先前的值进行检查。这在仅特定类的 UPDATE 对并发问题敏感的方案中可能很有用：

# will leave version_uuid unchanged
u1.name = "u3"
session.commit()

类映射 API

原文：docs.sqlalchemy.org/en/20/orm/mapping_api.html

class sqlalchemy.orm.registry

用于映射类的通用注册表。

registry 用作维护映射集合的基础，并提供用于映射类的配置钩子。

支持的三种常规映射类型是声明基类（Declarative Base）、声明装饰器（Declarative Decorator）和命令式映射（Imperative Mapping）。所有这些映射样式都可以互换使用：

• registry.generate_base() 返回一个新的声明基类，是 declarative_base() 函数的底层实现。
  
• registry.mapped() 提供了一个类装饰器，它将为一个类应用声明性映射，而不使用声明性基类。
  
• registry.map_imperatively() 会为一个类生成一个 Mapper，而不会扫描该类以寻找声明性类属性。这种方法适用于历史上由 sqlalchemy.orm.mapper() 传统映射函数提供的用例，该函数已在 SQLAlchemy 2.0 中移除。
  

从版本 1.4 新增。

成员

init(), as_declarative_base(), configure(),  dispose(), generate_base(), map_declaratively(), map_imperatively(),  mapped(), mapped_as_dataclass(), mappers, update_type_annotation_map()

参见

ORM 映射类概述 - 类映射样式概述。

method __init__(*, metadata: Optional[MetaData] = None, class_registry: Optional[clsregistry._ClsRegistryType] = None, type_annotation_map: Optional[_TypeAnnotationMapType] = None, constructor: Callable[..., None] = <function _declarative_constructor>)

构建一个新的 registry

参数：

• metadata – 一个可选的 MetaData 实例。使用声明性表映射生成的所有 Table 对象将使用此 MetaData 集合。如果将此参数保留在默认值 None，则会创建一个空白的 MetaData 集合。
  
• constructor – 指定在没有自己的 __init__ 的映射类上的 __init__ 函数的实现。默认情况下，为声明的字段和关系分配 **kwargs 的实现分配给一个实例。如果提供 None，则不会提供 init，并且构造将回退到 cls.init 的普通 Python 语义。
  
• class_registry – 可选的字典，当使用字符串名称来标识 relationship() 等内部类时，将充当类名称->映射类的注册表。允许两个或更多声明性基类共享相同的类名称注册表，以简化基类之间的关系。
  
• type_annotation_map –
  可选的 Python 类型到 SQLAlchemy TypeEngine类或实例的字典。提供的字典将更新默认类型映射。这仅由MappedColumn构造在Mapped类型内部的注解产生列类型时使用。
  新版本 2.0 中的内容。
  另请参阅
  自定义类型映射
  

method as_declarative_base(**kw: Any) → Callable[[Type[_T]], Type[_T]]

类装饰器，将为给定的基类调用registry.generate_base()。

例如：

from sqlalchemy.orm import registry
mapper_registry = registry()
@mapper_registry.as_declarative_base()
class Base:
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()
    id = Column(Integer, primary_key=True)
class MyMappedClass(Base):
    # ...

所有传递给registry.as_declarative_base()的关键字参数都会传递给registry.generate_base()。

method configure(cascade: bool = False) → None

配置此注册表中所有尚未配置的映射器。

配置步骤用于调和和初始化relationship()链接，以及调用配置事件，如MapperEvents.before_configured()和MapperEvents.after_configured()，这些事件可以被 ORM 扩展或用户定义的扩展钩子所使用。

如果此注册表中的一个或多个映射器包含指向其他注册表中映射类的relationship()构造，则称该注册表为依赖于那些注册表。为了自动配置这些依赖注册表，configure.cascade标志应设置为True。否则，如果它们未配置，则会引发异常。此行为背后的原理是允许应用程序在控制是否隐式到达其他注册表的同时，以编程方式调用注册表的配置。

作为调用registry.configure()的替代方案，可以使用 ORM 函数configure_mappers()函数确保内存中所有registry对象的配置完成。这通常更简单，并且还早于整体使用registry对象的用法。但是，此函数将影响运行中的 Python 进程中的所有映射，并且对于具有许多用于不同目的的注册表的应用程序可能更耗费内存/时间，这些注册表可能不会立即需要。

另请参阅

configure_mappers()

自版本 1.4.0b2 新增。

method dispose(cascade: bool = False) → None

处理此 registry 中的所有映射器。

调用后，此注册表中映射的所有类将不再具有与类相关联的类仪器。该方法是每个registry的类似于应用程序范围的clear_mappers()函数。

如果此注册表包含其他注册表的依赖项映射器，通常通过relationship()链接，则必须将这些注册表也处理掉。当这些注册表存在于与此相关的关系中时，如果设置了dispose.cascade标志为True，则它们的registry.dispose()方法也将被调用；否则，如果这些注册表尚未被处理，则会引发错误。

自版本 1.4.0b2 新增。

另请参阅

clear_mappers()

method generate_base(mapper: ~typing.Callable[[...], ~sqlalchemy.orm.mapper.Mapper[~typing.Any]] | None = None, cls: ~typing.Type[~typing.Any] = <class 'object'>, name: str = 'Base', metaclass: ~typing.Type[~typing.Any] = <class 'sqlalchemy.orm.decl_api.DeclarativeMeta'>) → Any

生成一个声明性基类。

继承自返回的类对象的类将使用声明性映射自动映射。

例如：

from sqlalchemy.orm import registry
mapper_registry = registry()
Base = mapper_registry.generate_base()
class MyClass(Base):
    __tablename__ = "my_table"
    id = Column(Integer, primary_key=True)

上述动态生成的类等同于下面的非动态示例：

from sqlalchemy.orm import registry
from sqlalchemy.orm.decl_api import DeclarativeMeta
mapper_registry = registry()
class Base(metaclass=DeclarativeMeta):
    __abstract__ = True
    registry = mapper_registry
    metadata = mapper_registry.metadata
    __init__ = mapper_registry.constructor

自版本 2.0 变更：请注意，registry.generate_base()方法已被新的DeclarativeBase类取代，该类使用子类化生成一个新的“基”类，而不是函数的返回值。这样可以与PEP 484类型工具兼容的方法。

registry.generate_base()方法提供了declarative_base()函数的实现，该函数一次性创建了registry和基类。

查看声明式映射部分以获取背景和示例。

参数：

• mapper – 可选可调用对象，默认为Mapper。此函数用于生成新的Mapper对象。
  
• cls – 默认为object。要用作生成的声明性基类的基础的类型。可以是类或类的元组。
  
• name – 默认为Base。生成类的显示名称。虽然不需要自定义此项，但可以提高回溯和调试时的清晰度。
  
• metaclass – 默认为DeclarativeMeta。作为生成的声明性基类的元类型的元类或__metaclass__兼容可调用对象。
  

另请参阅

声明式映射

declarative_base()

method map_declaratively(cls: Type[_O]) → Mapper[_O]

声明性地映射一个类。

在这种映射形式中，将扫描类以获取映射信息，包括要与表关联的列和/或实际表对象。

返回Mapper对象。

例如：

from sqlalchemy.orm import registry
mapper_registry = registry()
class Foo:
    __tablename__ = 'some_table'
    id = Column(Integer, primary_key=True)
    name = Column(String)
mapper = mapper_registry.map_declaratively(Foo)

此函数更方便地通过registry.mapped()类装饰器或通过从registry.generate_base()生成的声明性元类的子类间接调用。

查看完整详情和示例，请参阅声明式映射部分。

SqlAlchemy 2.0 中文文档（十）（3）https://developer.aliyun.com/article/1562917