数据库后端ORM【免费下载链接】sqlalchemyThe Database Toolkit for Python项目地址https://gitcode.com/gh_mirrors/sq/sqlalchemy点击查看免费下载导读SQLAlchemy ORM 默认只检测属性重新赋值无法感知字典、列表、集合等可变对象内部的原地修改in-place mutation。sqlalchemy.ext.mutable扩展通过Mutable、MutableComposite等混合类把这种内部变更透明地传播给父级映射对象使其进入Session.dirty集合并在 flush 时正确落库。本文以 doc/build/orm/extensions/mutable.rst 为核心骨架结合 lib/sqlalchemy/ext/mutable.py 的源码实现与 test/ext/test_mutable.py 的测试用例系统讲解标量类型JSON 字典、Pickle、ARRAY 等与 composite 复合属性的可变追踪原理、接入方式与序列化注意事项。读完你将能够为自定义TypeDecorator接入可变追踪、使用官方MutableDict/MutableList/MutableSet、在 composite 映射中启用原地修改检测并正确编写__getstate__/__setstate__以支持 pickling。一、问题背景为什么 ORM 检测不到原地修改SQLAlchemy ORM 的变更追踪change tracking建立在属性被重新赋值这一事件之上。当我们执行obj.json_value[key] value # 不会触发任何 set 事件时被修改的是json_value这个 Python 对象内部的数据结构对象的引用本身没有变化因此 ORM 的 instrumentation 无从得知数据已经改变——除非随后把整个值重新赋回去obj.json_value {key: value} # 触发 set 事件ORM 可以检测到这一限制在 doc/build/core/custom_types.rst 的 Adding Mutability 一节中有明确表述该节同时给出了接入MutableDict.as_mutable()的推荐做法。对于创建后不再修改的数据这种限制并无大碍但对于需要频繁原地更新 JSON、列表、集合或自定义复合对象的应用就必须借助sqlalchemy.ext.mutable扩展。二、核心设计值对象反向追踪父级sqlalchemy.ext.mutable的核心理念非常巧妙让被修改的值自己记住哪些父对象通过哪个属性名引用它一旦值发生内部变化就主动通知所有父对象。这个机制依赖MutableBase._parents属性——一个惰性初始化的weakref.WeakKeyDictionary其键为父对象的InstanceState值为父对象上的属性名。参见 lib/sqlalchemy/ext/mutable.py 中MutableBase的定义memoized_property def _parents(self) - WeakKeyDictionary[Any, Any]: Dictionary of parent objects InstanceState-attribute name on the parent. return weakref.WeakKeyDictionary()注意两点它使用weakref 弱引用因此不会阻止父对象被垃圾回收也不会形成引用环导致内存泄漏从 1.4 版本起字典的键从实例本身改为InstanceState见源码 docstring 中的.. versionchanged:: 1.4说明。当某个可变值对象发出变更信号时Mutable.changed()会遍历_parents对每个父对象调用flag_modified(parent.obj(), key)def changed(self) - None: Subclasses should call this method whenever change events occur. for parent, key in self._parents.items(): flag_modified(parent.obj(), key)而flag_modified定义于 lib/sqlalchemy/orm/attributes.py 的flag_modified函数会完成两件事把实例标记为 dirty并为该属性派发一次modified变更事件。这就是后面接收事件一节中AttributeEvents.modified能生效的底层原因。三、标量列上的可变追踪Mutable Scalar Values3.1 第一步自定义一个 JSON 编码的 TypeDecorator文档首先给出一个典型的 JSON 字典场景。我们自定义一个TypeDecorator把 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 端目标类型是可变的的 SQL 类型包括PickleType、PostgreSQL 的ARRAY等。源码 docstring 明确列出了这些候选lib/sqlalchemy/ext/mutable.py顶部说明测试 test/ext/test_mutable.py 中MutableColumnCopyArrayTest就用MutableList.as_mutable(ARRAY(Integer))验证了数组类型的场景。3.2 第二步编写 MutableDict 子类把变更路由到 changed()接下来我们把Mutable混入内建dict得到文档中的MutableDict示例类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()关键点有两个重写__setitem__/__delitem__在真正修改底层字典后调用self.changed()。这是每次原地修改都通知父级的钩子。文档同时指出实现思路不唯一——你也可以继承UserDict或collections.MutableMapping重要的是确保每次原地变更都触发changed()。重写coerce()类方法用于把不是MutableDict的值例如json模块返回的普通dict转换为目标类型。这一步是可选的——如果你的JSONEncodedDict始终返回MutableDict实例、并且调用方也保证传入MutableDict就可以不重写。一旦不重写coerce任何非目标类型的赋值都会抛出ValueError这正是MutableBase.coerce的默认行为lib/sqlalchemy/ext/mutable.py中coerce默认实现value is None时返回None否则抛出Attribute %s does not accept objects of type %s的ValueError。测试 test/ext/test_mutable.py 的test_coerce_raise就验证了这种报错路径。3.3 第三步用 as_mutable() 绑定列类型Mutable提供类方法as_mutable()它接收一个类型对象或类型类返回该类型的实例并注册一个监听器未来任何把该类型映射到 ORM 属性的操作都会被自动附加变更追踪 instrumentation。经典 Table 元数据Core 风格用法from sqlalchemy import Table, Column, Integer my_data Table( my_data, metadata, Column(id, Integer, primary_keyTrue), Column(data, MutableDict.as_mutable(JSONEncodedDict)), )Declarative 风格2.0 的Mapped/mapped_column用法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_keyTrue) data: Mapped[dict[str, str]] mapped_column( MutableDict.as_mutable(JSONEncodedDict) )映射完成后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测试 test/ext/test_mutable.py 中的test_in_place_mutation完整验证了这一行为f1.data[a] c之后提交数据库中的值确实被更新为{a: c}。as_mutable 的源码细节从源码lib/sqlalchemy/ext/mutable.py的Mutable.as_mutable可以看到它的两个重要设计始终返回类型实例即使传入的是类型类如PickleType也会通过types.to_instance()转为实例从而允许内联使用Column(data, MyMutableType.as_mutable(PickleType))SchemaType 的特殊处理如果类型是SchemaEventTarget如 JSON、ARRAYColumn 被复制时类型也会被复制会丢失与原始类型的关联。因此源码通过before_parent_attach事件在Column.info[_ext_mutable_orig_type]中记录原始类型mapper_configured时据此识别见as_mutable内的_add_column_memo与listen_for_type。这正是MutableColumnCopyJSONTest中基类/子类映射后 Column 类型不再是同一个对象但追踪依然生效的原因。只针对指定实例as_mutable只对明确使用了该类型实例的列附加 instrumentation不会无条件作用于该类型的所有出现。3.4 第四步用 associate_with() 全局关联类型如果希望所有JSONEncodedDict的出现都被MutableDict拦截无需逐个声明可以使用associate_with()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_keyTrue) data: Mapped[dict[str, str]] mapped_column(JSONEncodedDict)associate_with的实现lib/sqlalchemy/ext/mutable.py监听Mapper.mapper_configured事件遍历每个 mapper 的column_attrs凡是首列类型匹配sqltype的属性都自动调用associate_with_attribute()挂上监听。⚠️ 重要警告associate_with以及as_mutable注册的监听器是全局的、不会被垃圾回收的。文档与源码 docstring 双重警告只应将其用于应用生命周期内永久存在的类型不要用于临时/adhoc 类型否则会造成内存无界增长源码中associate_with与as_mutable的 docstring 均包含这一warning。此外还有一个更精确的入口associate_with_attribute()用于直接对某个已映射的描述符挂载监听。测试MutableAssocWithAttrInheritTesttest/ext/test_mutable.py展示了它配合继承映射Foo.data挂载后SubFoo也生效的用法。3.5 官方内置的三种可变容器文档对应的 API Reference 列出了三个官方实现均位于 lib/sqlalchemy/ext/mutable.py类底层类型覆盖的原地修改方法均触发changed()MutableDictdict__setitem__、__delitem__、setdefault、update、pop、popitem、clearMutableListlist__setitem__、__delitem__、pop、append、extend、__iadd__、insert、remove、clear、sort、reverseMutableSetsetupdate、intersection_update、difference_update、symmetric_difference_update、__ior__、__iand__、__ixor__、__isub__、add、remove、discard、pop、clear三个类都重写了coerce把普通dict/list/set转换为自身类型并实现了__getstate__/__setstate__见下文序列化一节。MutableList额外实现了__reduce_ex__以保证旧版 pickle 数据的向后兼容test/ext/test_mutable.py 的test_legacy_pickle_loads专门用若干历史 pickle 字节串验证了这一兼容性。深度提示这三个容器不会对容器内部的元素做递归的可变追踪。也就是说MutableDict无法完整解决嵌套 JSON 结构的深度变更追踪——例如data[a][b] 1这类二级修改修改的是内层普通dict外层MutableDict的__setitem__并不会被触发。官方文档明确建议如需递归追踪应自行继承MutableDict等类为放入容器内的值提供合适的 coercion使内层值本身也是可变的并能向上级结构派发事件。测试 test/ext/test_mutable.py 的test_in_place_mutation_strf1.data[1] [four, two]也证实了列表内元素被替换时外层能感知而元素内部再修改则需要递归方案。四、序列化支持Supporting Pickling4.1 为什么 _parents 不能进 picklesqlalchemy.ext.mutable的根基是挂在值对象上的weakref.WeakKeyDictionary即_parents。WeakKeyDictionary本身不可 pickle——它包含弱引用和函数回调。在官方看来这反而是好事如果这个字典可 pickle当值对象脱离父对象单独序列化时pickle 体积会异常膨胀因为里面挂着对父对象的弱引用结构。因此开发者的责任是提供一个__getstate__把_parents从 pickle 流中剔除class MyMutableType(Mutable): def __getstate__(self): d self.__dict__.copy() d.pop(_parents, None) return d对于字典示例还要同时序列化并恢复字典内容本身class MutableDict(Mutable, dict): # .... def __getstate__(self): return dict(self) def __setstate__(self, state): self.update(state)4.2 父对象一起 pickle 时如何重建 _parents文档特别说明当可变值对象是附着在一个或多个也参与 pickle 的父对象上被序列化时Mutable混合类会在父对象反序列化的过程中自动为每个值对象重建_parents集合。这个机制在源码中由_listen_on_attribute注册的pickle/unpickle事件处理器实现lib/sqlalchemy/ext/mutable.pypickle把当前值对象收集进state_dict[ext.mutable.values][key]unpickle父对象解包时将val._parents[state] key重新登记恢复值 → 父对象的反向索引。注意unpickle中还兼容了旧版格式collection为 list 的情况注释标注为 legacy format。测试 test/ext/test_mutable.py 的test_pickle_parent验证了父对象 pickle 后对data的原地修改依然能让对象进入sess.dirtytest_pickle_parent_multi_attrs对应 issue #8133则验证了同一父对象上挂多个 mutable 属性时pickle 往返后每个属性的 history 都正确。4.3 MutableList 的 pickle 兼容细节MutableList由于pickle与copy机制的不一致额外实现了__reduce_ex__返回(self.__class__, (list(self),))并保留__setstate__以兼容旧 pickle。测试test_legacy_pickle_loads中保留了 6 份历史 pickle 字节串确保新代码仍能反序列化旧格式。五、接收变更事件Receiving EventsMutable.changed()最终调用的是sqlalchemy.orm.attributes.flag_modified这会触发属性的modified事件。于是可以像监听普通 set 事件一样用AttributeEvents.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_keyTrue) 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)底层依据lib/sqlalchemy/orm/attributes.py的flag_modifiedstate, dict_ instance_state(instance), instance_dict(instance) impl state.manager[key].impl impl.dispatch.modified(state, impl._modified_token) state._modified_event(dict_, impl, NO_VALUE, is_userlandTrue)即先派发modified事件可被event.listens_for(..., modified)捕获再把实例标记为已修改进入Session.dirty。测试 test/ext/test_mutable.py 的test_modified_event用 mock 精确断言了f1.data[a] c会以(f1, AttributeEventToken(Foo.data.impl, OP_MODIFIED))的形式触发一次modified回调。六、Composites 上的可变追踪Mutable Composite Values6.1 什么是 compositeComposites 是 ORM 的一项特性一个标量属性对应底层表的一个或多个列组合成的对象。官方文档以几何学中的点Point为例详见 doc/build/orm/composites.rst 的mapper_composite一节。在 2.0 中推荐用 dataclass 配合composite()构造器使用。6.2 让 Point 支持原地修改继承 MutableComposite与Mutable类似用户定义的 composite 类以MutableComposite作为混入并通过MutableComposite.changed()把变更派发给父级。对 composite 类而言检测点通常落在特殊方法__setattr__上。文档在mapper_composite的Point基础上扩展如下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通过类映射事件mapper_configured自动为所有使用Point类型的composite()建立监听。模块加载时_setup_composite_listener()会注册全局监听lib/sqlalchemy/ext/mutable.py模块末尾只要composite_class是MutableComposite的子类就自动调用_listen_on_attribute。因此下面的Vertex映射无需额外声明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_keyTrue) start: Mapped[Point] composite( mapped_column(x1), mapped_column(y1) ) end: Mapped[Point] composite( mapped_column(x2), mapped_column(y2) ) def __repr__(self): return fVertex(start{self.start}, end{self.end})Point对象上的任何原地修改都会标记父对象为 dirty。文档给出的完整会话演示含 SQL 输出 from sqlalchemy.orm import Session sess Session(engine) v1 Vertex(startPoint(3, 4), endPoint(12, 15)) sess.add(v1) 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 sess.commit() UPDATE vertices SET x2? WHERE vertices.id ? [...] (8, 1) COMMIT注意输出中UPDATE只更新了x2一列——这正是MutableComposite.changed()的实现效果。从源码看changed()会把 composite 拆回各列的值并逐一写回父对象的底层列属性def changed(self) - None: for parent, key in self._parents.items(): prop parent.mapper.get_property(key) for value, attr_name in zip( prop._composite_values_from_instance(self), prop._attribute_keys, ): setattr(parent.obj(), attr_name, value)这也解释了MutableComposite._get_listen_keys为什么要在attribute.key之外把 composite 涉及的底层列属性键attribute.property._attribute_keys一并纳入——这样 refresh/refresh_flush 等事件才能正确感知与 composite 关联的列是否被刷新lib/sqlalchemy/ext/mutable.py的MutableComposite._get_listen_keys。6.3 复合类型的 coercionMutableBase.coerce同样支持 composite 类型但调用时机与标量不同对MutableComposite而言coerce只在属性 set 操作时被调用加载load时不会调用——加载时的类型转换由composite()构造器机制本身处理源码 docstring 对此有明确说明。覆盖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测试 test/ext/test_mutable.py 中的MyPoint与MyDCPoint提供了同样的模式把元组 coerce 成Point。注意Point类上若未定义__eq__测试中会额外实现以支持断言dataclass 默认生成的__eq__即可满足需求。6.4 Composite 的 pickling 支持与Mutable一致MutableComposite也依赖不可 pickle 的WeakKeyDictionary_parents。如果要 picklePoint或其宿主类Vertex至少要提供不包含_parents的__getstate__。文档给出的最小化实现把 composite 打包成元组dataclasses.dataclass class Point(MutableComposite): # ... def __getstate__(self): return self.x, self.y def __setstate__(self, state): self.x, self.y state与Mutable相同MutableComposite也会在父对象反序列化时自动恢复每个Point对象的_parents集合源码中_listen_on_attribute注册的pickle/unpickle事件同样适用于 composite 场景。七、事件挂载机制的源码级剖析无论是标量还是 composite所有追踪能力最终都汇聚在MutableBase._listen_on_attribute()lib/sqlalchemy/ext/mutable.py。它为指定映射属性注册一组事件事件处理器作用_sa_event_merge_wo_loadloadmerge 且不加载时包装值loadload对象加载时coerce值并登记val._parents[state] keyrefresh/refresh_flushload_attrs刷新时若涉及监听键则重新登记依赖_get_listen_keys属性setrawTrue, retvalTrueset_赋值时coerce 新值、登记新值到父级、解除旧值的登记oldvalue._parents.pop(inspect(target), None)picklepickle把值对象收集进ext.mutable.valuesunpickleunpickle反序列化时重建_parentsset_处理器是理解替换 vs 原地修改的关键当value is oldvalue时直接返回避免自赋值触发无谓事件测试test_replace_itself_still_ok验证了f1.data f1.data后原地修改依然正常当赋入的值不是目标类型时调用cls.coerce(key, value)转换随后为新值登记父级、为旧值解除登记。这也正是普通dict赋值给MutableDict属性会被自动 coerce 成MutableDict的机制来源测试test_replacef1.data {b: c}后提交数据正确持久化。load处理器同时解释了文档中coerce对Mutable在 set 和 load 时都会调用、对MutableComposite只在 set 时调用的差异_listen_on_attribute的coerce参数由调用方决定——associate_with_attribute/as_mutable/associate_with传入True标量场景 load 时也要 coerce而_setup_composite_listener传入Falsecomposite 的 load 转换交给composite()自身。八、可验证的测试用例清单如果你希望在本仓库中亲手验证上述行为test/ext/test_mutable.py 是权威参考值得关注的关键用例包括test_in_place_mutation/test_update/test_clear/test_pop/test_setdefaultMutableDict各方法的原地修改均能进入 dirty 并正确提交test_modified_event用 mock 精确断言modified事件的触发签名test_replace/test_replace_itself_still_ok赋值替换与自赋值的行为边界test_pickle_parent/test_pickle_parent_multi_attrspickle 往返后追踪仍然生效issue #8133test_legacy_pickle_loadsMutableList对历史 pickle 格式的兼容MutableWithScalarJSONTest.test_non_mutable对照组——未接入 mutable 的JSONEncodedDict列原地修改不会被检测f1.non_mutable_data[a] c提交后值不变MutableColumnDefaultTest.test_evt_on_flush_refresh列默认值default{}在 flush/refresh 后也是MutableDict实例issue #3427test_no_duplicate_reg_w_inheritance继承映射下 set 事件监听不会重复注册issue #9676test_session_mergetransient/detached 对象 merge 后原地修改依然产生正确的 historyissue #8446。运行方式仓库根目录下遵循 README.unittests.rst 的测试说明python -m pytest test/ext/test_mutable.py -v九、实战要点速查接入标量可变类型的三步编写/选用可变值类型如MutableDict→ 用as_mutable(sqltype)或associate_with(sqltype)绑定 SQL 类型 → 正常映射即可接入 composite 可变类型composite 类继承MutableComposite 在__setattr__中调用changed()映射层无需额外声明模块导入即自动生效coerce 语义Mutable在 set 与 load 时都会 coerceMutableComposite只在 set 时 coerce等价于validatespickling务必提供剔除_parents的__getstate__以及配套__setstate__父对象整体 pickle 时_parents会自动重建监听变更通过event.listens_for(Cls.attr, modified)捕获flag_modified派发的事件内存警告as_mutable/associate_with注册的监听器全局且不可回收只用于应用级永久类型递归结构官方容器不做深度追踪嵌套 JSON 等场景需要自定义值也是 mutable的递归容器适用类型范围任何 Python 端目标可变的 SQL 类型PickleType、PostgreSQLARRAY、JSON 等均可接入不限于 JSON 字典。十、总结sqlalchemy.ext.mutable用值反向登记父级_parents→ 变更时逐级广播changed()→flag_modified的设计把 ORM 的变更追踪从属性赋值级推进到容器/复合对象内部修改级。理解其coerce语义、pickling 约束与全局监听器内存代价是正确使用Mutable、MutableComposite以及官方MutableDict/MutableList/MutableSet的前提。本文涉及的完整源码、API 参考与测试用例可分别查阅 lib/sqlalchemy/ext/mutable.py、doc/build/orm/extensions/mutable.rst、doc/build/orm/composites.rst 与 test/ext/test_mutable.py。赞分享数据库后端ORM【免费下载链接】sqlalchemyThe Database Toolkit for Python项目地址https://gitcode.com/gh_mirrors/sq/sqlalchemy点击查看免费下载相关推荐SQLAlchemy 0.7 迁移指南新事件系统、混合属性与可变性追踪全面解析SQLAlchemy 0.7 迁移指南新事件系统、混合属性与可变性追踪全面解析 本篇指南基于 SQLAlchemy 官方迁移文档 doc/build/cha数据库后端ORM如何快速部署DeltaForce-OBS-Locker从零开始的OBS锁头插件完整指南如何快速部署DeltaForce OBS Locker从零开始的OBS锁头插件完整指南 DeltaForce OBS Locker是一款基于OBS渲染注入的智计算机视觉如何用biliTickerBuy轻松抢到B站热门门票终极快速入门指南如何用biliTickerBuy轻松抢到B站热门门票终极快速入门指南 还在为抢不到心仪的B站会员购门票而烦恼吗每次热门演唱会、漫展门票开售时手速总是不够快数据库后端ORM上一篇Hayase离线观看功能深度使用随时随地享受已下载的动漫下一篇解锁Windows性能AtlasOS故障容忍堆(FTH)深度优化指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
