变异跟踪

为跟踪标量值的原地更改提供支持，这些更改会传播到拥有父对象的 ORM 更改事件中。

在标量列值上建立可变性

“可变”结构的典型示例是 Python 字典。按照 SQL 数据类型对象 中介绍的示例，我们从一个自定义类型开始，该类型在持久化之前将 Python 字典编组为 JSON 字符串。

from sqlalchemy.types import TypeDecorator, VARCHAR
import json

class JSONEncodedDict(TypeDecorator):
    "Represents an immutable structure as a json-encoded string."

    impl = VARCHAR

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

json 的使用仅用于示例。 sqlalchemy.ext.mutable 扩展可以与任何其目标 Python 类型可能是可变的类型一起使用，包括 PickleType、ARRAY 等。

使用 sqlalchemy.ext.mutable 扩展时，值本身会跟踪所有引用它的父级。下面，我们展示了 MutableDict 字典对象的简单版本，该版本将 Mutable 混合到一个普通的 Python 字典中。

from sqlalchemy.ext.mutable import Mutable

class MutableDict(Mutable, dict):
    @classmethod
    def coerce(cls, key, value):
        "Convert plain dictionaries to MutableDict."

        if not isinstance(value, MutableDict):
            if isinstance(value, dict):
                return MutableDict(value)

            # this call will raise ValueError
            return Mutable.coerce(key, value)
        else:
            return value

    def __setitem__(self, key, value):
        "Detect dictionary set events and emit change events."

        dict.__setitem__(self, key, value)
        self.changed()

    def __delitem__(self, key):
        "Detect dictionary del events and emit change events."

        dict.__delitem__(self, key)
        self.changed()

上面的字典类采用子类化 Python 内置 dict 的方法来生成一个 dict 子类，该子类通过 __setitem__ 传递所有变异事件。这种方法有几种变体，例如子类化 UserDict.UserDict 或 collections.MutableMapping；对本示例来说，重要的是在数据结构发生原地更改时，会调用 Mutable.changed() 方法。

我们还重新定义了 Mutable.coerce() 方法，该方法将用于将所有不是 MutableDict 实例的值（例如 json 模块返回的普通字典）转换为适当的类型。定义此方法是可选的；我们也可以创建我们的 JSONEncodedDict，使其始终返回 MutableDict 的实例，并确保所有调用代码都显式地使用 MutableDict。当 Mutable.coerce() 未被重写时，应用于父对象的任何不是可变类型实例的值都会引发 ValueError。

我们的新 MutableDict 类型提供了一个类方法 Mutable.as_mutable()，我们可以在列元数据中使用它来与类型相关联。此方法获取给定的类型对象或类，并关联一个侦听器，该侦听器将检测该类型的未来所有映射，并对映射的属性应用事件侦听机制。例如，使用经典的表元数据

from sqlalchemy import Table, Column, Integer

my_data = Table('my_data', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', MutableDict.as_mutable(JSONEncodedDict))
)

上面，Mutable.as_mutable() 返回一个 JSONEncodedDict 的实例（如果类型对象不是实例），它将拦截所有针对此类型映射的属性。下面，我们针对 my_data 表建立一个简单的映射

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

class Base(DeclarativeBase):
    pass

class MyDataClass(Base):
    __tablename__ = 'my_data'
    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))

MyDataClass.data 成员现在将收到其值原地更改的通知。

对 MyDataClass.data 成员的任何原地更改将标记父对象上的属性为“脏”。

>>> from sqlalchemy.orm import Session

>>> sess = Session(some_engine)
>>> m1 = MyDataClass(data={'value1':'foo'})
>>> sess.add(m1)
>>> sess.commit()

>>> m1.data['value1'] = 'bar'
>>> assert m1 in sess.dirty
True

MutableDict 可以使用 Mutable.associate_with() 与所有未来的 JSONEncodedDict 实例相关联。这与 Mutable.as_mutable() 类似，只是它将无条件地拦截所有映射中 MutableDict 的所有出现，而无需单独声明。

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

MutableDict.associate_with(JSONEncodedDict)

class Base(DeclarativeBase):
    pass

class MyDataClass(Base):
    __tablename__ = 'my_data'
    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[dict[str, str]] = mapped_column(JSONEncodedDict)

支持序列化

sqlalchemy.ext.mutable 扩展的关键在于将一个 weakref.WeakKeyDictionary 放置在值对象上，该对象存储父映射对象的映射，这些映射对象以它们与该值关联的属性名称为键。 WeakKeyDictionary 对象不可序列化，因为它们包含弱引用和函数回调。在我们的例子中，这是一件好事，因为如果这个字典是可序列化的，它会导致在父对象上下文之外独立序列化的值对象产生过大的序列化大小。开发人员的责任仅限于提供一个 __getstate__ 方法，该方法将 MutableBase._parents() 集合从序列化流中排除

class MyMutableType(Mutable):
    def __getstate__(self):
        d = self.__dict__.copy()
        d.pop('_parents', None)
        return d

在我们的字典示例中，我们需要返回字典本身的内容（并在 __setstate__ 上恢复它们）

class MutableDict(Mutable, dict):
    # ....

    def __getstate__(self):
        return dict(self)

    def __setstate__(self, state):
        self.update(state)

在我们的可变值对象与其作为序列化的一部分的多个父对象相连的情况下， Mutable 混合将为每个值对象重新建立 Mutable._parents 集合，因为拥有父对象本身被反序列化了。

接收事件

当可变标量发出更改事件时，可以使用 AttributeEvents.modified() 事件处理程序来接收事件。当从可变扩展中调用 flag_modified() 函数时，会调用此事件处理程序。

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

class Base(DeclarativeBase):
    pass

class MyDataClass(Base):
    __tablename__ = 'my_data'
    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))

@event.listens_for(MyDataClass.data, "modified")
def modified_json(instance, initiator):
    print("json value modified:", instance.data)

在复合类型上建立可变性

复合类型是 ORM 的一项特殊功能，允许将单个标量属性分配一个对象值，该值代表从底层映射表中一个或多个列“组合”的信息。常见的例子是几何“点”，在 复合列类型 中介绍。

与 Mutable 一样，用户定义的复合类型类将 MutableComposite 作为混入子类化，并通过 MutableComposite.changed() 方法检测和向其父级传递更改事件。对于复合类，检测通常通过使用特殊的 Python 方法 __setattr__() 来完成。在下面的示例中，我们扩展了 复合列类型 中介绍的 Point 类，使其包含 MutableComposite 作为其基类，并通过 __setattr__ 将属性设置事件路由到 MutableComposite.changed() 方法。

import dataclasses
from sqlalchemy.ext.mutable import MutableComposite

@dataclasses.dataclass
class Point(MutableComposite):
    x: int
    y: int

    def __setattr__(self, key, value):
        "Intercept set events"

        # set the attribute
        object.__setattr__(self, key, value)

        # alert all parents to the change
        self.changed()

MutableComposite 类使用类映射事件来自动为任何使用 composite() 指定我们的 Point 类型的操作建立监听器。在下文中，当 Point 映射到 Vertex 类时，将建立监听器，这些监听器会将来自 Point 对象的更改事件路由到每个 Vertex.start 和 Vertex.end 属性。

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

class Base(DeclarativeBase):
    pass


class Vertex(Base):
    __tablename__ = "vertices"

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

    start: Mapped[Point] = composite(mapped_column("x1"), mapped_column("y1"))
    end: Mapped[Point] = composite(mapped_column("x2"), mapped_column("y2"))

    def __repr__(self):
        return f"Vertex(start={self.start}, end={self.end})"

对 Vertex.start 或 Vertex.end 成员的任何就地更改都会将该属性标记为“脏”状态，并将其应用到父对象。

>>> from sqlalchemy.orm import Session
>>> sess = Session(engine)
>>> v1 = Vertex(start=Point(3, 4), end=Point(12, 15))
>>> sess.add(v1)
sql>>> sess.flush()
BEGIN (implicit)
INSERT INTO vertices (x1, y1, x2, y2) VALUES (?, ?, ?, ?)
[...] (3, 4, 12, 15)
>>> v1.end.x = 8
>>> assert v1 in sess.dirty
True
sql>>> sess.commit()
UPDATE vertices SET x2=? WHERE vertices.id = ?
[...] (8, 1)
COMMIT

强制转换可变复合类型

MutableBase.coerce() 方法也支持在复合类型上使用。对于 MutableComposite，MutableBase.coerce() 方法仅针对属性设置操作调用，不针对加载操作调用。覆盖 MutableBase.coerce() 方法等效于对使用自定义复合类型的所有属性使用 validates() 验证例程。

@dataclasses.dataclass
class Point(MutableComposite):
    # other Point methods
    # ...

    def coerce(cls, key, value):
        if isinstance(value, tuple):
            value = Point(*value)
        elif not isinstance(value, Point):
            raise ValueError("tuple or Point expected")
        return value

支持序列化

与 Mutable 一样，MutableComposite 帮助程序类使用一个 weakref.WeakKeyDictionary，该字典可通过 MutableBase._parents() 属性获得，该属性无法序列化。如果我们需要序列化 Point 或其拥有类 Vertex 的实例，我们至少需要定义一个 __getstate__，该定义不包含 _parents 字典。在下面，我们定义了 __getstate__ 和 __setstate__，它们封装了我们 Point 类的最小形式。

@dataclasses.dataclass
class Point(MutableComposite):
    # ...

    def __getstate__(self):
        return self.x, self.y

    def __setstate__(self, state):
        self.x, self.y = state

与 Mutable 一样，MutableComposite 扩展了父级对象-关系状态的序列化过程，以便将 MutableBase._parents() 集合恢复到所有 Point 对象中。

API 参考

对象名称	描述
Mutable	混入，定义将更改事件透明地传播到父对象。
MutableBase	Mutable 和 MutableComposite 的通用基类。
MutableComposite	混入，定义将 SQLAlchemy “复合”对象上的更改事件透明地传播到其拥有父级或父级。
MutableDict	实现 Mutable 的字典类型。
MutableList	实现 Mutable 的列表类型。
MutableSet	实现 Mutable 的集合类型。

class sqlalchemy.ext.mutable.MutableBase

成员

_parents, coerce()

Mutable 和 MutableComposite 的通用基类。

attribute sqlalchemy.ext.mutable.MutableBase._parents

父对象 InstanceState->父级上属性名称的字典。

此属性是一个所谓的“记忆化”属性。它在首次访问时使用新的 weakref.WeakKeyDictionary 初始化自身，在后续访问时返回同一个对象。

在版本 1.4 中更改: 现在使用 InstanceState 作为弱字典中的键，而不是实例本身。

classmethod sqlalchemy.ext.mutable.MutableBase.coerce(key: str, value: Any) → Any | None

给定一个值，将其强制转换为目标类型。

可以通过自定义子类覆盖，以将传入的数据强制转换为特定类型。

默认情况下，会引发 ValueError。

此方法在不同的场景中调用，具体取决于父类是 Mutable 类型还是 MutableComposite 类型。如果是前者，它将在属性设置操作和 ORM 加载操作期间被调用。对于后者，它只在属性设置操作期间被调用；composite() 结构的机制在加载操作期间处理强制转换。

参数:

• key – 正在设置的 ORM 映射属性的字符串名称。

• value – 传入的值。

返回值:

此方法应返回强制转换后的值，或者如果无法完成强制转换，则引发 ValueError。

class sqlalchemy.ext.mutable.Mutable

混入，定义将更改事件透明地传播到父对象。

有关用法信息，请参见 在标量列值上建立可变性 中的示例。

成员

_get_listen_keys(), _listen_on_attribute(), _parents, as_mutable(), associate_with(), associate_with_attribute(), changed(), coerce()

类签名

class sqlalchemy.ext.mutable.Mutable (sqlalchemy.ext.mutable.MutableBase)

classmethod sqlalchemy.ext.mutable.Mutable._get_listen_keys(attribute: QueryableAttribute[Any]) → Set[str]

继承自 sqlalchemy.ext.mutable.MutableBase._get_listen_keys 方法 MutableBase

给定一个描述符属性，返回一个 set()，其中包含指示此属性状态发生变化的属性键。

这通常只是 set([attribute.key])，但可以覆盖以提供其他键。例如，MutableComposite 使用与构成复合值的列相关的属性键来扩充此集合。

此集合是在拦截 InstanceEvents.refresh() 和 InstanceEvents.refresh_flush() 事件的情况下进行查询的，这些事件将传递一个已刷新属性名称列表；该列表与该集合进行比较以确定是否需要采取措施。

classmethod sqlalchemy.ext.mutable.Mutable._listen_on_attribute(attribute: QueryableAttribute[Any], coerce: bool, parent_cls: _ExternalEntityType[Any]) → None

继承自 sqlalchemy.ext.mutable.MutableBase._listen_on_attribute 方法 MutableBase

将此类型设置为给定映射描述符的变异监听器。

attribute sqlalchemy.ext.mutable.Mutable._parents

继承自 sqlalchemy.ext.mutable.MutableBase._parents 属性 MutableBase

父对象 InstanceState->父级上属性名称的字典。

此属性是一个所谓的“记忆化”属性。它在首次访问时使用新的 weakref.WeakKeyDictionary 初始化自身，在后续访问时返回同一个对象。

在版本 1.4 中更改: 现在使用 InstanceState 作为弱字典中的键，而不是实例本身。

classmethod sqlalchemy.ext.mutable.Mutable.as_mutable(sqltype: _TypeEngineArgument[_T]) → TypeEngine[_T]

将 SQL 类型与该可变 Python 类型相关联。

这将建立监听器，这些监听器将检测针对给定类型的 ORM 映射，并将变异事件跟踪器添加到这些映射中。

该类型将无条件地作为实例返回，以便 as_mutable() 可以内联使用。

Table('mytable', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', MyMutableType.as_mutable(PickleType))
)

请注意，返回的类型始终是一个实例，即使给定一个类，并且只有专门使用该类型实例声明的列才会收到额外的检测。

若要将特定可变类型与特定类型的全部出现相关联，请使用特定 Mutable 子类的 Mutable.associate_with() 类方法以建立全局关联。

警告

此方法建立的监听器对所有映射器都是全局的，并且不会被垃圾回收。仅对应用程序中永久存在的类型使用 as_mutable()，而不是对临时类型使用，否则会导致内存使用量无限制地增长。

classmethod sqlalchemy.ext.mutable.Mutable.associate_with(sqltype: type) → None

将此包装器与将来映射的给定类型的所有列相关联。

这是一个便利方法，它会自动调用 associate_with_attribute。

警告

此方法建立的监听器对所有映射器都是全局的，并且不会被垃圾回收。仅对应用程序中永久存在的类型使用 associate_with()，而不是对临时类型使用，否则会导致内存使用量无限制地增长。

classmethod sqlalchemy.ext.mutable.Mutable.associate_with_attribute(attribute: InstrumentedAttribute[_O]) → None

将此类型设置为给定映射描述符的变异监听器。

method sqlalchemy.ext.mutable.Mutable.changed() → None

子类在发生变更事件时应该调用此方法。

classmethod sqlalchemy.ext.mutable.Mutable.coerce(key: str, value: Any) → Any | None

继承自 MutableBase.coerce() 方法 MutableBase

给定一个值，将其强制转换为目标类型。

可以通过自定义子类覆盖，以将传入的数据强制转换为特定类型。

默认情况下，会引发 ValueError。

此方法在不同的场景中调用，具体取决于父类是 Mutable 类型还是 MutableComposite 类型。如果是前者，它将在属性设置操作和 ORM 加载操作期间被调用。对于后者，它只在属性设置操作期间被调用；composite() 结构的机制在加载操作期间处理强制转换。

参数:

• key – ORM 映射属性的字符串名称。

• value – 接收到的值。

返回值:

此方法应返回强制转换后的值，或者如果无法完成强制转换，则引发 ValueError。

class sqlalchemy.ext.mutable.MutableComposite

混入，定义将 SQLAlchemy “复合”对象上的更改事件透明地传播到其拥有父级或父级。

有关使用方法的信息，请参阅 在复合类型上建立可变性 中的示例。

成员

changed()

类签名

class sqlalchemy.ext.mutable.MutableComposite (sqlalchemy.ext.mutable.MutableBase)

method sqlalchemy.ext.mutable.MutableComposite.changed() → None

子类在发生变更事件时应该调用此方法。

class sqlalchemy.ext.mutable.MutableDict

实现 Mutable 的字典类型。

MutableDict 对象实现了字典，该字典会在字典内容更改时（包括添加或删除值时）向基础映射发出更改事件。

请注意，MutableDict 不会对字典内部的值本身应用可变跟踪。因此，对于跟踪对递归字典结构（例如 JSON 结构）的深层更改的用例来说，它不是一个足够的解决方案。要支持此用例，请构建一个 MutableDict 的子类，该子类为放置在字典中的值提供适当的强制转换，使其也变为“可变的”，并向上级结构发出事件。

另请参阅

MutableList

MutableSet

成员

clear(), coerce(), pop(), popitem(), setdefault(), update()

类签名

class sqlalchemy.ext.mutable.MutableDict (sqlalchemy.ext.mutable.Mutable, builtins.dict, typing.Generic)

method sqlalchemy.ext.mutable.MutableDict.clear() → None.  Remove all items from D.

classmethod sqlalchemy.ext.mutable.MutableDict.coerce(key: str, value: Any) → MutableDict[_KT, _VT] | None

将普通字典转换为此类的实例。

method sqlalchemy.ext.mutable.MutableDict.pop(k[, d]) → v, remove specified key and return the corresponding value.

如果未找到键，则返回默认值（如果已给出）；否则，将引发 KeyError。

method sqlalchemy.ext.mutable.MutableDict.popitem() → Tuple[_KT, _VT]

删除并返回一个 (key, value) 对作为 2 元组。

对按 LIFO（后进先出）顺序返回。如果字典为空，则引发 KeyError。

method sqlalchemy.ext.mutable.MutableDict.setdefault(*arg)

如果键不在字典中，则使用默认值的键插入。

如果键在字典中，则返回键的值，否则返回默认值。

method sqlalchemy.ext.mutable.MutableDict.update([E, ]**F) → None.  Update D from dict/iterable E and F.

如果 E 存在且具有 .keys() 方法，则执行：for k in E: D[k] = E[k] 如果 E 存在且缺少 .keys() 方法，则执行：for k, v in E: D[k] = v 在任一情况下，之后都会执行：for k in F: D[k] = F[k]

class sqlalchemy.ext.mutable.MutableList

实现 Mutable 的列表类型。

MutableList 对象实现了一个列表，该列表会在列表内容更改时（包括添加或删除值时）向基础映射发出更改事件。

请注意，MutableList 不会对列表内部的值本身应用可变跟踪。因此，对于跟踪对递归可变结构（例如 JSON 结构）的深层更改的用例来说，它不是一个足够的解决方案。要支持此用例，请构建一个 MutableList 的子类，该子类为放置在字典中的值提供适当的强制转换，使其也变为“可变的”，并向上级结构发出事件。

另请参阅

MutableDict

MutableSet

成员

append()，clear()，coerce()，extend()，insert()，is_iterable()，is_scalar()，pop()，remove()，reverse()，sort()

类签名

class sqlalchemy.ext.mutable.MutableList (sqlalchemy.ext.mutable.Mutable，builtins.list，typing.Generic)

method sqlalchemy.ext.mutable.MutableList.append(x: _T) → None

将对象追加到列表末尾。

method sqlalchemy.ext.mutable.MutableList.clear() → None

从列表中删除所有项目。

classmethod sqlalchemy.ext.mutable.MutableList.coerce(key: str, value: MutableList[_T] | _T) → MutableList[_T] | None

将普通列表转换为该类的实例。

method sqlalchemy.ext.mutable.MutableList.extend(x: Iterable[_T]) → None

通过将来自可迭代对象的元素追加到列表中来扩展列表。

method sqlalchemy.ext.mutable.MutableList.insert(i: SupportsIndex, x: _T) → None

在索引之前插入对象。

method sqlalchemy.ext.mutable.MutableList.is_iterable(value: _T | Iterable[_T]) → TypeGuard[Iterable[_T]]

method sqlalchemy.ext.mutable.MutableList.is_scalar(value: _T | Iterable[_T]) → TypeGuard[_T]

method sqlalchemy.ext.mutable.MutableList.pop(*arg: SupportsIndex) → _T

删除并返回索引处的项目（默认情况下为最后一个）。

如果列表为空或索引超出范围，则引发 IndexError。

method sqlalchemy.ext.mutable.MutableList.remove(i: _T) → None

删除值的第一次出现。

如果值不存在，则引发 ValueError。

method sqlalchemy.ext.mutable.MutableList.reverse() → None

反转 *就地*。

method sqlalchemy.ext.mutable.MutableList.sort(**kw: Any) → None

按升序对列表进行排序并返回 None。

排序是就地排序（即列表本身被修改）并且稳定（即两个相等元素的顺序保持不变）。

如果给出了键函数，则将其应用于每个列表项一次，并根据它们的函数值按升序或降序对它们进行排序。

reverse 标志可以设置为按降序排序。

class sqlalchemy.ext.mutable.MutableSet

实现 Mutable 的集合类型。

MutableSet 对象实现了一个集合，当集合的内容发生更改时（包括添加或删除值时），它将向底层映射发出更改事件。

请注意，MutableSet 不会 对集合内部的 *值本身* 应用可变跟踪。因此，它不适用于跟踪对 *递归* 可变结构的深度更改的用例。要支持此用例，请构建 MutableSet 的子类，该子类对放置在字典中的值提供适当的强制转换，以使它们也“可变”，并向其父结构发出事件。

另请参阅

MutableDict

MutableList

成员

add(), clear(), coerce(), difference_update(), discard(), intersection_update(), pop(), remove(), symmetric_difference_update(), update()

类签名

类 sqlalchemy.ext.mutable.MutableSet (sqlalchemy.ext.mutable.Mutable, builtins.set, typing.Generic)

方法 sqlalchemy.ext.mutable.MutableSet.add(elem: _T) → None

将元素添加到集合中。

如果元素已存在，则此操作无效。

方法 sqlalchemy.ext.mutable.MutableSet.clear() → None

从该集合中移除所有元素。

类方法 sqlalchemy.ext.mutable.MutableSet.coerce(index: str, value: Any) → MutableSet[_T] | None

将普通集合转换为该类的实例。

方法 sqlalchemy.ext.mutable.MutableSet.difference_update(*arg: Iterable[Any]) → None

从该集合中移除另一个集合中的所有元素。

方法 sqlalchemy.ext.mutable.MutableSet.discard(elem: _T) → None

如果元素是集合的成员，则将其从集合中移除。

如果元素不是成员，则不执行任何操作。

方法 sqlalchemy.ext.mutable.MutableSet.intersection_update(*arg: Iterable[Any]) → None

使用自身与另一个集合的交集更新集合。

方法 sqlalchemy.ext.mutable.MutableSet.pop(*arg: Any) → _T

移除并返回一个任意的集合元素。如果集合为空，则会引发 KeyError。

方法 sqlalchemy.ext.mutable.MutableSet.remove(elem: _T) → None

从集合中移除一个元素；它必须是成员。

如果元素不是成员，则会引发 KeyError。

方法 sqlalchemy.ext.mutable.MutableSet.symmetric_difference_update(*arg: Iterable[_T]) → None

使用自身与另一个集合的对称差更新集合。

方法 sqlalchemy.ext.mutable.MutableSet.update(*arg: Iterable[_T]) → None

使用自身与其他集合的并集更新集合。