突变跟踪


支持跟踪对标量值的就地更改,这些更改将传播到拥有父对象上的 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 类型可能是可变的任何类型的 Python 类型,包括 PickleTypeARRAY 等。


当使用 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()


上面的 dictionary 类采用将 Python 内置 dict 子类化的方法,以生成一个 dict 子类,该子类通过 __setitem__ 路由所有 mutation 事件。这种方法有一些变体,例如子类化 UserDict.UserDict收集。MutableMapping的 Mapping;这个例子很重要的部分是,每当对数据结构进行就地更改时,都会调用 Mutable.changed() 方法。


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


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

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


MutableDict 可以通过一个步骤与 JSONEncodedDict 的所有未来实例相关联,使用 Mutable.associate_with() 中。 这类似于 Mutable.as_mutable() 不同之处在于它将无条件地拦截所有 mappings 中出现的所有 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)


支持 Pickling


sqlalchemy.ext.mutable 扩展的关键依赖于 weakref 的位置。WeakKeyDictionary 的 value 对象,它存储父映射对象的映射,该映射对象与它们与此值关联的属性名称作为键。WeakKeyDictionary 对象是不可 picklable 的,因为它们包含 weakref 和函数回调。在我们的例子中,这是一件好事,因为如果这个字典是可 picklable 的,它可能会导致我们的值对象变得过大,这些对象在父级的上下文之外被自己 pickle 大小。这里的开发人员责任只是提供一个 __getstate__ 方法,该方法从 pickle 流中排除 MutableBase._parents() 集合:

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


在我们的字典示例中,我们需要返回 dict 本身的内容(并在 __setstate__ 上恢复它们):

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

    def __getstate__(self):
        return dict(self)

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


如果我们的可变值对象被腌制,因为它被附加到一个或多个父对象,而这些父对象也是 pickle 的一部分,则 Mutable Mixin 将在每个 value 对象上重新建立 Mutable._parents 集合,因为拥有父对象本身被解封。


接收事件


AttributeEvents.modified() 事件处理程序可用于在可变标量发出 change 事件时接收事件。当从可变扩展中调用 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 功能,它允许为单个标量属性分配一个对象值,该值表示从底层映射表中的一个或多个列“组成”的信息。通常的示例是几何 “点” 的示例,在 复合列类型 中介绍。


Thousand 一样,用户定义的复合类将 MutableComposite 作为 mixin 子类化,并通过 MutableComposite.changed() 方法检测更改事件并将其传递给其父类。对于复合类,通常通过使用特殊的 Python 方法 __setattr__() 进行检测。在下面的示例中,我们扩展了 PointComposite Column Types 中引入的类包括 MutableComposite 的 c,并通过 __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() method 仅对属性集作调用,而不对 load作调用。 重写 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


支持 Pickling


Mutable 的情况一样,MutableComposite 帮助程序类使用 weakref。WeakKeyDictionaryMutableBase._parents() 属性。如果我们需要 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


Multable 一样,MutableComposite 增强了 Pickling 过程,以便 MutableBase._parents() 集合将还原到所有 Point 对象。


API 参考


对象名称

描述


可变


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

MutableBase


Mutable 的公共基类 和 MutableComposite 的 Composite 实例。

MutableComposite


Mixin,它定义了 SQLAlchemy “复合” 对象上更改事件到其拥有的父级或父级的透明传播。


MutableDict 函数


实现 Variable 的字典类型。

MutableList


实现 Mutable 的列表类型。

MutableSet


实现 Mutable 的 set 类型。


sqlalchemy.ext.mutable。MutableBase


Mutable 的公共基类 和 MutableComposite 的 Composite 实例。


属性 sqlalchemy.ext.mutable.MutableBase. _parents


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


这个属性是所谓的 “memoized” 属性。它使用新的 weakref 初始化自身。WeakKeyDictionary 的调用,在后续访问时返回相同的对象。


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


classmethod sqlalchemy.ext.mutable.MutableBase. coercekey strvalue Any→AnyNone


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


可以被自定义子类覆盖,以将传入数据强制转换为特定类型。


默认情况下,会引发 ValueError


此方法在不同场景中调用,具体取决于父类是 Mutable 类型还是 类型 MutableComposite 的在前者的情况下,attribute-set作和 ORM 加载作都会调用它。对于后者,它仅在属性集作期间调用;composite() 构造的机制处理加载作期间的强制转换。


参数

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


  • value —— 传入的值。


结果


该方法应返回强制值,或引发 ValueError 表示无法完成强制转换。


sqlalchemy.ext.mutable。可变的¶


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


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


类方法 sqlalchemy.ext.mutable.Mutable. _get_listen_keysattribute QueryableAttribute[Any] Set[str]


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


给定一个 descriptor 属性,返回一个属性键的 set(),该键指示此属性状态的更改。


这通常只是 set([attribute.key]),但可以被覆盖以提供额外的键。例如 MutableComposite 使用与列关联的属性键来扩充此集 ,这些值构成 Composite 值。


在拦截 InstanceEvents.refresh() InstanceEvents.refresh_flush() 事件,传递已刷新的属性名称列表;将列表与此集进行比较,以确定是否需要采取措施。


类方法 sqlalchemy.ext.mutable.Mutable. _listen_on_attributeattribute QueryableAttribute[Any]强制 boolparent_cls: _ExternalEntityType[任意]


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


将此类型建立为给定 Map Descriptors 的 mutation 侦听器。


属性 sqlalchemy.ext.mutable.Mutable. _parents


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


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


这个属性是所谓的 “memoized” 属性。它使用新的 weakref 初始化自身。WeakKeyDictionary 的调用,在后续访问时返回相同的对象。


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


类方法 sqlalchemy.ext.mutable.Mutable. as_mutableSQLTatype _TypeEngineArgument[_T] TypeEngine[_T]


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


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


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

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


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


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


警告


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


类方法 sqlalchemy.ext.mutable.Mutable. associate_withSQLtype type


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


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


警告


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


类方法 sqlalchemy.ext.mutable.Mutable. associate_with_attributeattribute InstrumentedAttribute[_O]


将此类型建立为给定 Map Descriptors 的 mutation 侦听器。


method sqlalchemy.ext.mutable.Mutable. changed


每当更改事件发生时,子类都应该调用此方法。


classmethod sqlalchemy.ext.mutable.Mutable. coercekey strvalue Any→AnyNone


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


可以被自定义子类覆盖,以将传入数据强制转换为特定类型。


默认情况下,会引发 ValueError


此方法在不同场景中调用,具体取决于父类是 Mutable 类型还是 类型 MutableComposite 的在前者的情况下,attribute-set作和 ORM 加载作都会调用它。对于后者,它仅在属性集作期间调用;composite() 构造的机制处理加载作期间的强制转换。


参数

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


  • value —— 传入的值。


结果


该方法应返回强制值,或引发 ValueError 表示无法完成强制转换。


sqlalchemy.ext.mutable。MutableComposite


Mixin,它定义了 SQLAlchemy “复合” 对象上更改事件到其拥有的父级或父级的透明传播。


有关使用信息,请参阅在复合上建立可变性中的示例。


method sqlalchemy.ext.mutable.MutableComposite. changed


每当更改事件发生时,子类都应该调用此方法。


sqlalchemy.ext.mutable。MutableDict 函数¶


实现 Variable 的字典类型。


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


请注意,MutableDict 不会将可变跟踪应用于 值本身。因此,对于跟踪递归的深度更改的用例来说,它不是一个足够的解决方案 字典结构,例如 JSON 结构。 为了支持此用例, 构建 MutableDict 的子类,该子类为字典中的值提供适当的强制转换,以便它们也是“可变的”,并将事件发送到其父结构。


另请参阅

MutableList

MutableSet


类签名


class sqlalchemy.ext.mutable.MutableDict sqlalchemy.ext.mutable.Mutable builtins.dicttyping.通用


方法 sqlalchemy.ext.mutable.MutableDict. clear None。  从 D 中删除所有项目

classmethod sqlalchemy.ext.mutable.MutableDict. coercekey strvalue Any→MutableDict[_KT,_VT]None(无)


将 plain dictionary 转换为此类的实例。


方法 sqlalchemy.ext.mutable.MutableDict. popk[d] v,删除指定的 key 返回 相应的值。


如果未找到 key,则返回 default(如果给定);否则,引发 KeyError。


方法 sqlalchemy.ext.mutable.MutableDict. popitem Tuple[_KT _VT]


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


对以 LIFO (后进先出) 顺序返回。如果 dict 为空,则引发 KeyError。


方法 sqlalchemy.ext.mutable.MutableDict. setDefault*arg


如果 key 不在字典中,则插入值为 default 的 key。


如果 key 在字典中,则返回 key 的值,否则返回 default。


method sqlalchemy.ext.mutable.MutableDict. update[E]**F None.  dict/iterable E F 更新 D


如果 E 存在并且有一个 .keys() 方法,则执行: 对于 E 中的 k: D[k] = E[k] 如果 E 存在并且缺少 .keys() 方法,则执行: 对于 k,E 中的 v: D[k] = v 在任何一种情况下,后跟: 对于 F 中的 k: D[k] = F[k]


sqlalchemy.ext.mutable。MutableList(可变列表)¶


实现 Mutable 的列表类型。


MutableList 对象实现一个列表,当列表的内容发生更改时,包括添加或删除值时,该列表将向底层映射发出更改事件。


请注意,MutableList 不会将可变跟踪应用于 值本身。因此,对于跟踪递归的深度更改的用例来说,它不是一个足够的解决方案 可变结构,例如 JSON 结构。 为了支持此用例, 构建 MutableList 的子类,该子类为字典中的值提供适当的强制转换,以便它们也是“可变的”,并将事件发送到其父结构。


类签名


class sqlalchemy.ext.mutable.MutableList sqlalchemy.ext.mutable.Mutable builtins.listtyping.通用


方法 sqlalchemy.ext.mutable.MutableList. appendx _T


将 object 附加到列表的末尾。


方法 sqlalchemy.ext.mutable.MutableList. clear


从列表中删除所有项目。


classmethod sqlalchemy.ext.mutable.MutableList. coercekey str, value:MutableList[_T]_T→MutableList[_T]无


将 plain list 转换为此类的实例。


method sqlalchemy.ext.mutable.MutableList. extendx Iterable[_T]


通过附加 iterable 中的元素来扩展 list。


方法 sqlalchemy.ext.mutable.MutableList. inserti SupportsIndexx _T None


在索引之前插入对象。


方法 sqlalchemy.ext.mutable.MutableList. is_iterablevalue:_TIterable[_T] TypeGuard[Iterable[_T]]

方法 sqlalchemy.ext.mutable.MutableList. is_scalarvalue:_TIterable[_T] TypeGuard[_T]

方法 sqlalchemy.ext.mutable.MutableList. pop*arg SupportsIndex _T


删除并返回 index 处的项目(默认为 last)。


如果 list 为空或 index 超出范围,则引发 IndexError 。


方法 sqlalchemy.ext.mutable.MutableList. removei _T None


删除第一个出现的值。


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


方法 sqlalchemy.ext.mutable.MutableList. reverse


反转 IN PLACE


method sqlalchemy.ext.mutable.MutableList. sort**kw Any None


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


排序是就地的(即列表本身被修改)和稳定的(即保持两个相等元素的顺序)。


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


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


sqlalchemy.ext.mutable。MutableSet(可变集


实现 Mutable 的 set 类型。


MutableSet 对象实现了一个 set,当 set 的内容发生更改时,包括添加或删除值时,该 set 将向底层映射发出 change 事件。


请注意,MutableSet 不会将可变跟踪应用于 值本身。因此,对于跟踪递归的深度更改的用例来说,它不是一个足够的解决方案 可变结构。 为了支持此用例, 构建一个 MutableSet 的子类,该子类为字典中的值提供适当的强制转换,以便它们也是“可变的”,并将事件发送到其父结构。


类签名


class sqlalchemy.ext.mutable.MutableSet sqlalchemy.ext.mutable.Mutable builtins.settyping.通用


method sqlalchemy.ext.mutable.MutableSet. addelem _T None(无)¶


将元素添加到集合中。


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


方法 sqlalchemy.ext.mutable.MutableSet. clear


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


classmethod sqlalchemy.ext.mutable.MutableSet. coerceindex strvalue Any→MutableSet[_T]无

Convert plain set to instance of this class.


method sqlalchemy.ext.mutable.MutableSet. difference_update*arg Iterable[Any]


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


方法 sqlalchemy.ext.mutable.MutableSet. discardelem _T None


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


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


method sqlalchemy.ext.mutable.MutableSet. intersection_update*arg Iterable[Any]


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


方法 sqlalchemy.ext.mutable.MutableSet. pop*arg Any _T


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


方法 sqlalchemy.ext.mutable.MutableSet. removeelem _T


从集合中删除元素;它必须是成员。


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


method sqlalchemy.ext.mutable.MutableSet. symmetric_difference_update*arg Iterable[_T]


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


method sqlalchemy.ext.mutable.MutableSet. update*arg Iterable[_T]


使用 itself 和 others 的并集更新集合。