Mutation Tracking (突变跟踪)

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

在标量列值上建立可变性

一个典型的 “mutable (可变)” 结构的例子是 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 mixin 应用于普通的 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__ 路由所有 mutation (突变) 事件。 这种方法有多种变体,例如子类化 UserDict.UserDictcollections.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 成员的任何就地更改都将把父对象上的属性标记为 “dirty (脏)”

>>> 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

可以使用 Mutable.associate_with() 一步将 MutableDictJSONEncodedDict 的所有未来实例关联。 这类似于 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)

支持 Pickle 序列化

sqlalchemy.ext.mutable 扩展的关键依赖于在值对象上放置 weakref.WeakKeyDictionary,它存储父映射对象的映射,键是它们与此值关联的属性名称。 WeakKeyDictionary 对象是不可 pickle 化的,因为它们包含 weakref 和函数回调。 在我们的例子中,这是一件好事,因为如果这个字典是可 pickle 化的,它可能会导致我们的值对象(在父对象的上下文之外被 pickle 化)的 pickle 大小过大。 开发人员的责任仅在于提供一个 __getstate__ 方法,该方法从 pickle 流中排除 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)

如果我们的可变值对象在附加到一个或多个也是 pickle 一部分的父对象时被 pickle 化,则当拥有父对象本身被 unpickle 化时,Mutable mixin 将在每个值对象上重新建立 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)

在复合类型上建立可变性

Composites (复合类型) 是一种特殊的 ORM 功能,它允许将单个标量属性分配一个对象值,该对象值表示从底层映射表中的一个或多个列 “composed (组成)” 的信息。 通常的例子是几何 “point (点)”,并在 复合列类型 中介绍。

Mutable 的情况一样,用户定义的 composite (复合) 类子类化 MutableComposite 作为 mixin,并通过 MutableComposite.changed() 方法检测更改事件并将其传递给其父对象。 在 composite (复合) 类的情况下,检测通常通过使用特殊的 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 类使用类映射事件来自动为任何指定我们 Point 类型的 composite() 的用法建立侦听器。 下面,当 Point 映射到 Vertex 类时,将建立侦听器,这些侦听器会将来自 Point 对象的更改事件路由到每个 Vertex.startVertex.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.startVertex.end 成员的任何就地更改都将把父对象上的属性标记为 “dirty (脏)”

>>> from sqlalchemy.orm import Session
>>> sess = Session(engine)
>>> v1 = Vertex(start=Point(3, 4), end=Point(12, 15))
>>> sess.add(v1)
sql>>> sess.flush()
>>> v1.end.x = 8
>>> assert v1 in sess.dirty
True
sql>>> sess.commit()

强制转换可变复合类型

MutableBase.coerce() 方法在复合类型上也受支持。 在 MutableComposite 的情况下,MutableBase.coerce() 方法仅针对属性设置操作调用,而不是加载操作。 重写 MutableBase.coerce() 方法本质上等同于为所有使用自定义 composite (复合) 类型的属性使用 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

支持 Pickle 序列化

Mutable 的情况一样,MutableComposite 辅助类使用通过 MutableBase._parents() 属性提供的 weakref.WeakKeyDictionary,它是不可 pickle 化的。 如果我们需要 pickle Point 或其拥有类 Vertex 的实例,我们至少需要定义一个不包含 _parents 字典的 __getstate__。 下面我们定义了 __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 增强了父对象关系状态的 pickle 过程,以便将 MutableBase._parents() 集合恢复到所有 Point 对象。

API 参考

对象名称 描述

Mutable

定义透明地将更改事件传播到父对象的 Mixin。

MutableBase

MutableMutableComposite 的通用基类。

MutableComposite

定义透明地将 SQLAlchemy “composite (复合)” 对象上的更改事件传播到其拥有父对象或父对象的 Mixin。

MutableDict

实现 Mutable 的字典类型。

MutableList

实现 Mutable 的列表类型。

MutableSet

实现 Mutable 的集合类型。

class sqlalchemy.ext.mutable.MutableBase

成员

_parents, coerce()

MutableMutableComposite 的通用基类。

attribute sqlalchemy.ext.mutable.MutableBase._parents

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

此属性是所谓的 “memoized (记忆化)” 属性。 它在首次访问时使用新的 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 – 传入的值。

Returns:

该方法应返回强制转换后的值,如果无法完成强制转换,则引发 ValueError

class sqlalchemy.ext.mutable.Mutable

定义透明地将更改事件传播到父对象的 Mixin。

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

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 使用与构成 composite (复合) 值的列关联的属性键来扩充此集合。

在拦截 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

将此类型建立为给定映射描述符的 mutation (突变) 侦听器。

attribute sqlalchemy.ext.mutable.Mutable._parents

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

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

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

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

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

将 SQL 类型与此 mutable (可变) Python 类型关联。

这建立侦听器,这些侦听器将检测针对给定类型的 ORM 映射,并将 mutation (突变) 事件跟踪器添加到这些映射。

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

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

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

要将特定的 mutable (可变) 类型与特定类型的所有出现关联,请使用特定 Mutable 子类的 Mutable.associate_with() 类方法来建立全局关联。

警告

由此方法建立的侦听器对于所有 mapper 都是全局的,并且不会被垃圾回收。 仅对应用程序永久存在的类型使用 as_mutable(),而不是 ad-hoc 类型,否则会导致内存使用量无限增长。

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

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

这是一个方便的方法,可以自动调用 associate_with_attribute

警告

由此方法建立的侦听器对于所有 mapper 都是全局的,并且不会被垃圾回收。 仅对应用程序永久存在的类型使用 associate_with(),而不是 ad-hoc 类型,否则会导致内存使用量无限增长。

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

将此类型建立为给定映射描述符的 mutation (突变) 侦听器。

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 – 传入的值。

Returns:

该方法应返回强制转换后的值,如果无法完成强制转换,则引发 ValueError

class sqlalchemy.ext.mutable.MutableComposite

定义透明地将 SQLAlchemy “composite (复合)” 对象上的更改事件传播到其拥有父对象或父对象的 Mixin。

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

成员

changed()

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

子类应在任何时候发生更改事件时调用此方法。

class sqlalchemy.ext.mutable.MutableDict

实现 Mutable 的字典类型。

MutableDict 对象实现了一个字典,当字典的内容被更改时(包括添加或删除值时),它将向底层映射发出更改事件。

请注意,MutableDict 对字典内部的值本身应用可变性跟踪。 因此,对于跟踪递归字典结构(例如 JSON 结构)的深层更改的用例来说,它不是一个充分的解决方案。 为了支持此用例,请构建 MutableDict 的子类,该子类为放入字典中的值提供适当的强制转换,以便它们也变成“可变的”,并将事件向上发出到其父结构。

另请参阅

MutableList

MutableSet

类签名

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]

删除并返回一个 (键, 值) 对,作为 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

类签名

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

类签名

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

method sqlalchemy.ext.mutable.MutableSet.add(elem: _T) None

向集合添加一个元素。

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

method sqlalchemy.ext.mutable.MutableSet.clear() None

从此集合中删除所有元素。

classmethod sqlalchemy.ext.mutable.MutableSet.coerce(index: str, value: Any) MutableSet[_T] | None

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

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

从此集合中删除另一个集合的所有元素。

method sqlalchemy.ext.mutable.MutableSet.discard(elem: _T) None

如果元素是成员,则从集合中删除该元素。

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

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

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

method sqlalchemy.ext.mutable.MutableSet.pop(*arg: Any) _T

删除并返回任意集合元素。如果集合为空,则引发 KeyError。

method 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

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