Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# ext/mutable.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8r"""Provide support for tracking of in-place changes to scalar values,
9which are propagated into ORM change events on owning parent objects.
11.. _mutable_scalars:
13Establishing Mutability on Scalar Column Values
14===============================================
16A typical example of a "mutable" structure is a Python dictionary.
17Following the example introduced in :ref:`types_toplevel`, we
18begin with a custom type that marshals Python dictionaries into
19JSON strings before being persisted::
21 from sqlalchemy.types import TypeDecorator, VARCHAR
22 import json
25 class JSONEncodedDict(TypeDecorator):
26 "Represents an immutable structure as a json-encoded string."
28 impl = VARCHAR
30 def process_bind_param(self, value, dialect):
31 if value is not None:
32 value = json.dumps(value)
33 return value
35 def process_result_value(self, value, dialect):
36 if value is not None:
37 value = json.loads(value)
38 return value
40The usage of ``json`` is only for the purposes of example. The
41:mod:`sqlalchemy.ext.mutable` extension can be used
42with any type whose target Python type may be mutable, including
43:class:`.PickleType`, :class:`_postgresql.ARRAY`, etc.
45When using the :mod:`sqlalchemy.ext.mutable` extension, the value itself
46tracks all parents which reference it. Below, we illustrate a simple
47version of the :class:`.MutableDict` dictionary object, which applies
48the :class:`.Mutable` mixin to a plain Python dictionary::
50 from sqlalchemy.ext.mutable import Mutable
53 class MutableDict(Mutable, dict):
54 @classmethod
55 def coerce(cls, key, value):
56 "Convert plain dictionaries to MutableDict."
58 if not isinstance(value, MutableDict):
59 if isinstance(value, dict):
60 return MutableDict(value)
62 # this call will raise ValueError
63 return Mutable.coerce(key, value)
64 else:
65 return value
67 def __setitem__(self, key, value):
68 "Detect dictionary set events and emit change events."
70 dict.__setitem__(self, key, value)
71 self.changed()
73 def __delitem__(self, key):
74 "Detect dictionary del events and emit change events."
76 dict.__delitem__(self, key)
77 self.changed()
79The above dictionary class takes the approach of subclassing the Python
80built-in ``dict`` to produce a dict
81subclass which routes all mutation events through ``__setitem__``. There are
82variants on this approach, such as subclassing ``UserDict.UserDict`` or
83``collections.MutableMapping``; the part that's important to this example is
84that the :meth:`.Mutable.changed` method is called whenever an in-place
85change to the datastructure takes place.
87We also redefine the :meth:`.Mutable.coerce` method which will be used to
88convert any values that are not instances of ``MutableDict``, such
89as the plain dictionaries returned by the ``json`` module, into the
90appropriate type. Defining this method is optional; we could just as well
91created our ``JSONEncodedDict`` such that it always returns an instance
92of ``MutableDict``, and additionally ensured that all calling code
93uses ``MutableDict`` explicitly. When :meth:`.Mutable.coerce` is not
94overridden, any values applied to a parent object which are not instances
95of the mutable type will raise a ``ValueError``.
97Our new ``MutableDict`` type offers a class method
98:meth:`~.Mutable.as_mutable` which we can use within column metadata
99to associate with types. This method grabs the given type object or
100class and associates a listener that will detect all future mappings
101of this type, applying event listening instrumentation to the mapped
102attribute. Such as, with classical table metadata::
104 from sqlalchemy import Table, Column, Integer
106 my_data = Table(
107 "my_data",
108 metadata,
109 Column("id", Integer, primary_key=True),
110 Column("data", MutableDict.as_mutable(JSONEncodedDict)),
111 )
113Above, :meth:`~.Mutable.as_mutable` returns an instance of ``JSONEncodedDict``
114(if the type object was not an instance already), which will intercept any
115attributes which are mapped against this type. Below we establish a simple
116mapping against the ``my_data`` table::
118 from sqlalchemy.orm import DeclarativeBase
119 from sqlalchemy.orm import Mapped
120 from sqlalchemy.orm import mapped_column
123 class Base(DeclarativeBase):
124 pass
127 class MyDataClass(Base):
128 __tablename__ = "my_data"
129 id: Mapped[int] = mapped_column(primary_key=True)
130 data: Mapped[dict[str, str]] = mapped_column(
131 MutableDict.as_mutable(JSONEncodedDict)
132 )
134The ``MyDataClass.data`` member will now be notified of in place changes
135to its value.
137Any in-place changes to the ``MyDataClass.data`` member
138will flag the attribute as "dirty" on the parent object::
140 >>> from sqlalchemy.orm import Session
142 >>> sess = Session(some_engine)
143 >>> m1 = MyDataClass(data={"value1": "foo"})
144 >>> sess.add(m1)
145 >>> sess.commit()
147 >>> m1.data["value1"] = "bar"
148 >>> assert m1 in sess.dirty
149 True
151The ``MutableDict`` can be associated with all future instances
152of ``JSONEncodedDict`` in one step, using
153:meth:`~.Mutable.associate_with`. This is similar to
154:meth:`~.Mutable.as_mutable` except it will intercept all occurrences
155of ``MutableDict`` in all mappings unconditionally, without
156the need to declare it individually::
158 from sqlalchemy.orm import DeclarativeBase
159 from sqlalchemy.orm import Mapped
160 from sqlalchemy.orm import mapped_column
162 MutableDict.associate_with(JSONEncodedDict)
165 class Base(DeclarativeBase):
166 pass
169 class MyDataClass(Base):
170 __tablename__ = "my_data"
171 id: Mapped[int] = mapped_column(primary_key=True)
172 data: Mapped[dict[str, str]] = mapped_column(JSONEncodedDict)
174Supporting Pickling
175--------------------
177The key to the :mod:`sqlalchemy.ext.mutable` extension relies upon the
178placement of a ``weakref.WeakKeyDictionary`` upon the value object, which
179stores a mapping of parent mapped objects keyed to the attribute name under
180which they are associated with this value. ``WeakKeyDictionary`` objects are
181not picklable, due to the fact that they contain weakrefs and function
182callbacks. In our case, this is a good thing, since if this dictionary were
183picklable, it could lead to an excessively large pickle size for our value
184objects that are pickled by themselves outside of the context of the parent.
185The developer responsibility here is only to provide a ``__getstate__`` method
186that excludes the :meth:`~MutableBase._parents` collection from the pickle
187stream::
189 class MyMutableType(Mutable):
190 def __getstate__(self):
191 d = self.__dict__.copy()
192 d.pop("_parents", None)
193 return d
195With our dictionary example, we need to return the contents of the dict itself
196(and also restore them on __setstate__)::
198 class MutableDict(Mutable, dict):
199 # ....
201 def __getstate__(self):
202 return dict(self)
204 def __setstate__(self, state):
205 self.update(state)
207In the case that our mutable value object is pickled as it is attached to one
208or more parent objects that are also part of the pickle, the :class:`.Mutable`
209mixin will re-establish the :attr:`.Mutable._parents` collection on each value
210object as the owning parents themselves are unpickled.
212Receiving Events
213----------------
215The :meth:`.AttributeEvents.modified` event handler may be used to receive
216an event when a mutable scalar emits a change event. This event handler
217is called when the :func:`.attributes.flag_modified` function is called
218from within the mutable extension::
220 from sqlalchemy.orm import DeclarativeBase
221 from sqlalchemy.orm import Mapped
222 from sqlalchemy.orm import mapped_column
223 from sqlalchemy import event
226 class Base(DeclarativeBase):
227 pass
230 class MyDataClass(Base):
231 __tablename__ = "my_data"
232 id: Mapped[int] = mapped_column(primary_key=True)
233 data: Mapped[dict[str, str]] = mapped_column(
234 MutableDict.as_mutable(JSONEncodedDict)
235 )
238 @event.listens_for(MyDataClass.data, "modified")
239 def modified_json(instance, initiator):
240 print("json value modified:", instance.data)
242.. _mutable_composites:
244Establishing Mutability on Composites
245=====================================
247Composites are a special ORM feature which allow a single scalar attribute to
248be assigned an object value which represents information "composed" from one
249or more columns from the underlying mapped table. The usual example is that of
250a geometric "point", and is introduced in :ref:`mapper_composite`.
252As is the case with :class:`.Mutable`, the user-defined composite class
253subclasses :class:`.MutableComposite` as a mixin, and detects and delivers
254change events to its parents via the :meth:`.MutableComposite.changed` method.
255In the case of a composite class, the detection is usually via the usage of the
256special Python method ``__setattr__()``. In the example below, we expand upon the ``Point``
257class introduced in :ref:`mapper_composite` to include
258:class:`.MutableComposite` in its bases and to route attribute set events via
259``__setattr__`` to the :meth:`.MutableComposite.changed` method::
261 import dataclasses
262 from sqlalchemy.ext.mutable import MutableComposite
265 @dataclasses.dataclass
266 class Point(MutableComposite):
267 x: int
268 y: int
270 def __setattr__(self, key, value):
271 "Intercept set events"
273 # set the attribute
274 object.__setattr__(self, key, value)
276 # alert all parents to the change
277 self.changed()
279The :class:`.MutableComposite` class makes use of class mapping events to
280automatically establish listeners for any usage of :func:`_orm.composite` that
281specifies our ``Point`` type. Below, when ``Point`` is mapped to the ``Vertex``
282class, listeners are established which will route change events from ``Point``
283objects to each of the ``Vertex.start`` and ``Vertex.end`` attributes::
285 from sqlalchemy.orm import DeclarativeBase, Mapped
286 from sqlalchemy.orm import composite, mapped_column
289 class Base(DeclarativeBase):
290 pass
293 class Vertex(Base):
294 __tablename__ = "vertices"
296 id: Mapped[int] = mapped_column(primary_key=True)
298 start: Mapped[Point] = composite(
299 mapped_column("x1"), mapped_column("y1")
300 )
301 end: Mapped[Point] = composite(
302 mapped_column("x2"), mapped_column("y2")
303 )
305 def __repr__(self):
306 return f"Vertex(start={self.start}, end={self.end})"
308Any in-place changes to the ``Vertex.start`` or ``Vertex.end`` members
309will flag the attribute as "dirty" on the parent object:
311.. sourcecode:: python+sql
313 >>> from sqlalchemy.orm import Session
314 >>> sess = Session(engine)
315 >>> v1 = Vertex(start=Point(3, 4), end=Point(12, 15))
316 >>> sess.add(v1)
317 {sql}>>> sess.flush()
318 BEGIN (implicit)
319 INSERT INTO vertices (x1, y1, x2, y2) VALUES (?, ?, ?, ?)
320 [...] (3, 4, 12, 15)
322 {stop}>>> v1.end.x = 8
323 >>> assert v1 in sess.dirty
324 True
325 {sql}>>> sess.commit()
326 UPDATE vertices SET x2=? WHERE vertices.id = ?
327 [...] (8, 1)
328 COMMIT
330Coercing Mutable Composites
331---------------------------
333The :meth:`.MutableBase.coerce` method is also supported on composite types.
334In the case of :class:`.MutableComposite`, the :meth:`.MutableBase.coerce`
335method is only called for attribute set operations, not load operations.
336Overriding the :meth:`.MutableBase.coerce` method is essentially equivalent
337to using a :func:`.validates` validation routine for all attributes which
338make use of the custom composite type::
340 @dataclasses.dataclass
341 class Point(MutableComposite):
342 # other Point methods
343 # ...
345 def coerce(cls, key, value):
346 if isinstance(value, tuple):
347 value = Point(*value)
348 elif not isinstance(value, Point):
349 raise ValueError("tuple or Point expected")
350 return value
352Supporting Pickling
353--------------------
355As is the case with :class:`.Mutable`, the :class:`.MutableComposite` helper
356class uses a ``weakref.WeakKeyDictionary`` available via the
357:meth:`MutableBase._parents` attribute which isn't picklable. If we need to
358pickle instances of ``Point`` or its owning class ``Vertex``, we at least need
359to define a ``__getstate__`` that doesn't include the ``_parents`` dictionary.
360Below we define both a ``__getstate__`` and a ``__setstate__`` that package up
361the minimal form of our ``Point`` class::
363 @dataclasses.dataclass
364 class Point(MutableComposite):
365 # ...
367 def __getstate__(self):
368 return self.x, self.y
370 def __setstate__(self, state):
371 self.x, self.y = state
373As with :class:`.Mutable`, the :class:`.MutableComposite` augments the
374pickling process of the parent's object-relational state so that the
375:meth:`MutableBase._parents` collection is restored to all ``Point`` objects.
377""" # noqa: E501
379from __future__ import annotations
381from collections import defaultdict
382from typing import AbstractSet
383from typing import Any
384from typing import Dict
385from typing import Iterable
386from typing import List
387from typing import Optional
388from typing import overload
389from typing import Set
390from typing import Tuple
391from typing import TYPE_CHECKING
392from typing import TypeVar
393from typing import Union
394import weakref
395from weakref import WeakKeyDictionary
397from .. import event
398from .. import inspect
399from .. import types
400from .. import util
401from ..orm import Mapper
402from ..orm._typing import _ExternalEntityType
403from ..orm._typing import _O
404from ..orm._typing import _T
405from ..orm.attributes import AttributeEventToken
406from ..orm.attributes import flag_modified
407from ..orm.attributes import InstrumentedAttribute
408from ..orm.attributes import QueryableAttribute
409from ..orm.context import QueryContext
410from ..orm.decl_api import DeclarativeAttributeIntercept
411from ..orm.state import InstanceState
412from ..orm.unitofwork import UOWTransaction
413from ..sql._typing import _TypeEngineArgument
414from ..sql.base import SchemaEventTarget
415from ..sql.schema import Column
416from ..sql.type_api import TypeEngine
417from ..util import memoized_property
418from ..util.typing import SupportsIndex
419from ..util.typing import TypeGuard
421_KT = TypeVar("_KT") # Key type.
422_VT = TypeVar("_VT") # Value type.
425class MutableBase:
426 """Common base class to :class:`.Mutable`
427 and :class:`.MutableComposite`.
429 """
431 @memoized_property
432 def _parents(self) -> WeakKeyDictionary[Any, Any]:
433 """Dictionary of parent object's :class:`.InstanceState`->attribute
434 name on the parent.
436 This attribute is a so-called "memoized" property. It initializes
437 itself with a new ``weakref.WeakKeyDictionary`` the first time
438 it is accessed, returning the same object upon subsequent access.
440 .. versionchanged:: 1.4 the :class:`.InstanceState` is now used
441 as the key in the weak dictionary rather than the instance
442 itself.
444 """
446 return weakref.WeakKeyDictionary()
448 @classmethod
449 def coerce(cls, key: str, value: Any) -> Optional[Any]:
450 """Given a value, coerce it into the target type.
452 Can be overridden by custom subclasses to coerce incoming
453 data into a particular type.
455 By default, raises ``ValueError``.
457 This method is called in different scenarios depending on if
458 the parent class is of type :class:`.Mutable` or of type
459 :class:`.MutableComposite`. In the case of the former, it is called
460 for both attribute-set operations as well as during ORM loading
461 operations. For the latter, it is only called during attribute-set
462 operations; the mechanics of the :func:`.composite` construct
463 handle coercion during load operations.
466 :param key: string name of the ORM-mapped attribute being set.
467 :param value: the incoming value.
468 :return: the method should return the coerced value, or raise
469 ``ValueError`` if the coercion cannot be completed.
471 """
472 if value is None:
473 return None
474 msg = "Attribute '%s' does not accept objects of type %s"
475 raise ValueError(msg % (key, type(value)))
477 @classmethod
478 def _get_listen_keys(cls, attribute: QueryableAttribute[Any]) -> Set[str]:
479 """Given a descriptor attribute, return a ``set()`` of the attribute
480 keys which indicate a change in the state of this attribute.
482 This is normally just ``set([attribute.key])``, but can be overridden
483 to provide for additional keys. E.g. a :class:`.MutableComposite`
484 augments this set with the attribute keys associated with the columns
485 that comprise the composite value.
487 This collection is consulted in the case of intercepting the
488 :meth:`.InstanceEvents.refresh` and
489 :meth:`.InstanceEvents.refresh_flush` events, which pass along a list
490 of attribute names that have been refreshed; the list is compared
491 against this set to determine if action needs to be taken.
493 """
494 return {attribute.key}
496 @classmethod
497 def _listen_on_attribute(
498 cls,
499 attribute: QueryableAttribute[Any],
500 coerce: bool,
501 parent_cls: _ExternalEntityType[Any],
502 ) -> None:
503 """Establish this type as a mutation listener for the given
504 mapped descriptor.
506 """
507 key = attribute.key
508 if parent_cls is not attribute.class_:
509 return
511 # rely on "propagate" here
512 parent_cls = attribute.class_
514 listen_keys = cls._get_listen_keys(attribute)
516 def load(state: InstanceState[_O], *args: Any) -> None:
517 """Listen for objects loaded or refreshed.
519 Wrap the target data member's value with
520 ``Mutable``.
522 """
523 val = state.dict.get(key, None)
524 if val is not None:
525 if coerce:
526 val = cls.coerce(key, val)
527 assert val is not None
528 state.dict[key] = val
529 val._parents[state] = key
531 def load_attrs(
532 state: InstanceState[_O],
533 ctx: Union[object, QueryContext, UOWTransaction],
534 attrs: Iterable[Any],
535 ) -> None:
536 if not attrs or listen_keys.intersection(attrs):
537 load(state)
539 def set_(
540 target: InstanceState[_O],
541 value: MutableBase | None,
542 oldvalue: MutableBase | None,
543 initiator: AttributeEventToken,
544 ) -> MutableBase | None:
545 """Listen for set/replace events on the target
546 data member.
548 Establish a weak reference to the parent object
549 on the incoming value, remove it for the one
550 outgoing.
552 """
553 if value is oldvalue:
554 return value
556 if not isinstance(value, cls):
557 value = cls.coerce(key, value)
558 if value is not None:
559 value._parents[target] = key
560 if isinstance(oldvalue, cls):
561 oldvalue._parents.pop(inspect(target), None)
562 return value
564 def pickle(
565 state: InstanceState[_O], state_dict: Dict[str, Any]
566 ) -> None:
567 val = state.dict.get(key, None)
568 if val is not None:
569 if "ext.mutable.values" not in state_dict:
570 state_dict["ext.mutable.values"] = defaultdict(list)
571 state_dict["ext.mutable.values"][key].append(val)
573 def unpickle(
574 state: InstanceState[_O], state_dict: Dict[str, Any]
575 ) -> None:
576 if "ext.mutable.values" in state_dict:
577 collection = state_dict["ext.mutable.values"]
578 if isinstance(collection, list):
579 # legacy format
580 for val in collection:
581 val._parents[state] = key
582 else:
583 for val in state_dict["ext.mutable.values"][key]:
584 val._parents[state] = key
586 event.listen(
587 parent_cls,
588 "_sa_event_merge_wo_load",
589 load,
590 raw=True,
591 propagate=True,
592 )
594 event.listen(parent_cls, "load", load, raw=True, propagate=True)
595 event.listen(
596 parent_cls, "refresh", load_attrs, raw=True, propagate=True
597 )
598 event.listen(
599 parent_cls, "refresh_flush", load_attrs, raw=True, propagate=True
600 )
601 event.listen(
602 attribute, "set", set_, raw=True, retval=True, propagate=True
603 )
604 event.listen(parent_cls, "pickle", pickle, raw=True, propagate=True)
605 event.listen(
606 parent_cls, "unpickle", unpickle, raw=True, propagate=True
607 )
610class Mutable(MutableBase):
611 """Mixin that defines transparent propagation of change
612 events to a parent object.
614 See the example in :ref:`mutable_scalars` for usage information.
616 """
618 def changed(self) -> None:
619 """Subclasses should call this method whenever change events occur."""
621 for parent, key in self._parents.items():
622 flag_modified(parent.obj(), key)
624 @classmethod
625 def associate_with_attribute(
626 cls, attribute: InstrumentedAttribute[_O]
627 ) -> None:
628 """Establish this type as a mutation listener for the given
629 mapped descriptor.
631 """
632 cls._listen_on_attribute(attribute, True, attribute.class_)
634 @classmethod
635 def associate_with(cls, sqltype: type) -> None:
636 """Associate this wrapper with all future mapped columns
637 of the given type.
639 This is a convenience method that calls
640 ``associate_with_attribute`` automatically.
642 .. warning::
644 The listeners established by this method are *global*
645 to all mappers, and are *not* garbage collected. Only use
646 :meth:`.associate_with` for types that are permanent to an
647 application, not with ad-hoc types else this will cause unbounded
648 growth in memory usage.
650 """
652 def listen_for_type(mapper: Mapper[_O], class_: type) -> None:
653 if mapper.non_primary:
654 return
655 for prop in mapper.column_attrs:
656 if isinstance(prop.columns[0].type, sqltype):
657 cls.associate_with_attribute(getattr(class_, prop.key))
659 event.listen(Mapper, "mapper_configured", listen_for_type)
661 @classmethod
662 def as_mutable(cls, sqltype: _TypeEngineArgument[_T]) -> TypeEngine[_T]:
663 """Associate a SQL type with this mutable Python type.
665 This establishes listeners that will detect ORM mappings against
666 the given type, adding mutation event trackers to those mappings.
668 The type is returned, unconditionally as an instance, so that
669 :meth:`.as_mutable` can be used inline::
671 Table(
672 "mytable",
673 metadata,
674 Column("id", Integer, primary_key=True),
675 Column("data", MyMutableType.as_mutable(PickleType)),
676 )
678 Note that the returned type is always an instance, even if a class
679 is given, and that only columns which are declared specifically with
680 that type instance receive additional instrumentation.
682 To associate a particular mutable type with all occurrences of a
683 particular type, use the :meth:`.Mutable.associate_with` classmethod
684 of the particular :class:`.Mutable` subclass to establish a global
685 association.
687 .. warning::
689 The listeners established by this method are *global*
690 to all mappers, and are *not* garbage collected. Only use
691 :meth:`.as_mutable` for types that are permanent to an application,
692 not with ad-hoc types else this will cause unbounded growth
693 in memory usage.
695 """
696 sqltype = types.to_instance(sqltype)
698 # a SchemaType will be copied when the Column is copied,
699 # and we'll lose our ability to link that type back to the original.
700 # so track our original type w/ columns
701 if isinstance(sqltype, SchemaEventTarget):
703 @event.listens_for(sqltype, "before_parent_attach")
704 def _add_column_memo(
705 sqltyp: TypeEngine[Any],
706 parent: Column[_T],
707 ) -> None:
708 parent.info["_ext_mutable_orig_type"] = sqltyp
710 schema_event_check = True
711 else:
712 schema_event_check = False
714 def listen_for_type(
715 mapper: Mapper[_T],
716 class_: Union[DeclarativeAttributeIntercept, type],
717 ) -> None:
718 if mapper.non_primary:
719 return
720 _APPLIED_KEY = "_ext_mutable_listener_applied"
722 for prop in mapper.column_attrs:
723 if (
724 # all Mutable types refer to a Column that's mapped,
725 # since this is the only kind of Core target the ORM can
726 # "mutate"
727 isinstance(prop.expression, Column)
728 and (
729 (
730 schema_event_check
731 and prop.expression.info.get(
732 "_ext_mutable_orig_type"
733 )
734 is sqltype
735 )
736 or prop.expression.type is sqltype
737 )
738 ):
739 if not prop.expression.info.get(_APPLIED_KEY, False):
740 prop.expression.info[_APPLIED_KEY] = True
741 cls.associate_with_attribute(getattr(class_, prop.key))
743 event.listen(Mapper, "mapper_configured", listen_for_type)
745 return sqltype
748class MutableComposite(MutableBase):
749 """Mixin that defines transparent propagation of change
750 events on a SQLAlchemy "composite" object to its
751 owning parent or parents.
753 See the example in :ref:`mutable_composites` for usage information.
755 """
757 @classmethod
758 def _get_listen_keys(cls, attribute: QueryableAttribute[_O]) -> Set[str]:
759 return {attribute.key}.union(attribute.property._attribute_keys)
761 def changed(self) -> None:
762 """Subclasses should call this method whenever change events occur."""
764 for parent, key in self._parents.items():
765 prop = parent.mapper.get_property(key)
766 for value, attr_name in zip(
767 prop._composite_values_from_instance(self),
768 prop._attribute_keys,
769 ):
770 setattr(parent.obj(), attr_name, value)
773def _setup_composite_listener() -> None:
774 def _listen_for_type(mapper: Mapper[_T], class_: type) -> None:
775 for prop in mapper.iterate_properties:
776 if (
777 hasattr(prop, "composite_class")
778 and isinstance(prop.composite_class, type)
779 and issubclass(prop.composite_class, MutableComposite)
780 ):
781 prop.composite_class._listen_on_attribute(
782 getattr(class_, prop.key), False, class_
783 )
785 if not event.contains(Mapper, "mapper_configured", _listen_for_type):
786 event.listen(Mapper, "mapper_configured", _listen_for_type)
789_setup_composite_listener()
792class MutableDict(Mutable, Dict[_KT, _VT]):
793 """A dictionary type that implements :class:`.Mutable`.
795 The :class:`.MutableDict` object implements a dictionary that will
796 emit change events to the underlying mapping when the contents of
797 the dictionary are altered, including when values are added or removed.
799 Note that :class:`.MutableDict` does **not** apply mutable tracking to the
800 *values themselves* inside the dictionary. Therefore it is not a sufficient
801 solution for the use case of tracking deep changes to a *recursive*
802 dictionary structure, such as a JSON structure. To support this use case,
803 build a subclass of :class:`.MutableDict` that provides appropriate
804 coercion to the values placed in the dictionary so that they too are
805 "mutable", and emit events up to their parent structure.
807 .. seealso::
809 :class:`.MutableList`
811 :class:`.MutableSet`
813 """
815 def __setitem__(self, key: _KT, value: _VT) -> None:
816 """Detect dictionary set events and emit change events."""
817 super().__setitem__(key, value)
818 self.changed()
820 if TYPE_CHECKING:
821 # from https://github.com/python/mypy/issues/14858
823 @overload
824 def setdefault(
825 self: MutableDict[_KT, Optional[_T]], key: _KT, value: None = None
826 ) -> Optional[_T]: ...
828 @overload
829 def setdefault(self, key: _KT, value: _VT) -> _VT: ...
831 def setdefault(self, key: _KT, value: object = None) -> object: ...
833 else:
835 def setdefault(self, *arg): # noqa: F811
836 result = super().setdefault(*arg)
837 self.changed()
838 return result
840 def __delitem__(self, key: _KT) -> None:
841 """Detect dictionary del events and emit change events."""
842 super().__delitem__(key)
843 self.changed()
845 def update(self, *a: Any, **kw: _VT) -> None:
846 super().update(*a, **kw)
847 self.changed()
849 if TYPE_CHECKING:
851 @overload
852 def pop(self, __key: _KT) -> _VT: ...
854 @overload
855 def pop(self, __key: _KT, __default: _VT | _T) -> _VT | _T: ...
857 def pop(
858 self, __key: _KT, __default: _VT | _T | None = None
859 ) -> _VT | _T: ...
861 else:
863 def pop(self, *arg): # noqa: F811
864 result = super().pop(*arg)
865 self.changed()
866 return result
868 def popitem(self) -> Tuple[_KT, _VT]:
869 result = super().popitem()
870 self.changed()
871 return result
873 def clear(self) -> None:
874 super().clear()
875 self.changed()
877 @classmethod
878 def coerce(cls, key: str, value: Any) -> MutableDict[_KT, _VT] | None:
879 """Convert plain dictionary to instance of this class."""
880 if not isinstance(value, cls):
881 if isinstance(value, dict):
882 return cls(value)
883 return Mutable.coerce(key, value)
884 else:
885 return value
887 def __getstate__(self) -> Dict[_KT, _VT]:
888 return dict(self)
890 def __setstate__(
891 self, state: Union[Dict[str, int], Dict[str, str]]
892 ) -> None:
893 self.update(state)
896class MutableList(Mutable, List[_T]):
897 """A list type that implements :class:`.Mutable`.
899 The :class:`.MutableList` object implements a list that will
900 emit change events to the underlying mapping when the contents of
901 the list are altered, including when values are added or removed.
903 Note that :class:`.MutableList` does **not** apply mutable tracking to the
904 *values themselves* inside the list. Therefore it is not a sufficient
905 solution for the use case of tracking deep changes to a *recursive*
906 mutable structure, such as a JSON structure. To support this use case,
907 build a subclass of :class:`.MutableList` that provides appropriate
908 coercion to the values placed in the dictionary so that they too are
909 "mutable", and emit events up to their parent structure.
911 .. seealso::
913 :class:`.MutableDict`
915 :class:`.MutableSet`
917 """
919 def __reduce_ex__(
920 self, proto: SupportsIndex
921 ) -> Tuple[type, Tuple[List[int]]]:
922 return (self.__class__, (list(self),))
924 # needed for backwards compatibility with
925 # older pickles
926 def __setstate__(self, state: Iterable[_T]) -> None:
927 self[:] = state
929 def is_scalar(self, value: _T | Iterable[_T]) -> TypeGuard[_T]:
930 return not util.is_non_string_iterable(value)
932 def is_iterable(self, value: _T | Iterable[_T]) -> TypeGuard[Iterable[_T]]:
933 return util.is_non_string_iterable(value)
935 def __setitem__(
936 self, index: SupportsIndex | slice, value: _T | Iterable[_T]
937 ) -> None:
938 """Detect list set events and emit change events."""
939 if isinstance(index, SupportsIndex) and self.is_scalar(value):
940 super().__setitem__(index, value)
941 elif isinstance(index, slice) and self.is_iterable(value):
942 super().__setitem__(index, value)
943 self.changed()
945 def __delitem__(self, index: SupportsIndex | slice) -> None:
946 """Detect list del events and emit change events."""
947 super().__delitem__(index)
948 self.changed()
950 def pop(self, *arg: SupportsIndex) -> _T:
951 result = super().pop(*arg)
952 self.changed()
953 return result
955 def append(self, x: _T) -> None:
956 super().append(x)
957 self.changed()
959 def extend(self, x: Iterable[_T]) -> None:
960 super().extend(x)
961 self.changed()
963 def __iadd__(self, x: Iterable[_T]) -> MutableList[_T]: # type: ignore[override,misc] # noqa: E501
964 self.extend(x)
965 return self
967 def insert(self, i: SupportsIndex, x: _T) -> None:
968 super().insert(i, x)
969 self.changed()
971 def remove(self, i: _T) -> None:
972 super().remove(i)
973 self.changed()
975 def clear(self) -> None:
976 super().clear()
977 self.changed()
979 def sort(self, **kw: Any) -> None:
980 super().sort(**kw)
981 self.changed()
983 def reverse(self) -> None:
984 super().reverse()
985 self.changed()
987 @classmethod
988 def coerce(
989 cls, key: str, value: MutableList[_T] | _T
990 ) -> Optional[MutableList[_T]]:
991 """Convert plain list to instance of this class."""
992 if not isinstance(value, cls):
993 if isinstance(value, list):
994 return cls(value)
995 return Mutable.coerce(key, value)
996 else:
997 return value
1000class MutableSet(Mutable, Set[_T]):
1001 """A set type that implements :class:`.Mutable`.
1003 The :class:`.MutableSet` object implements a set that will
1004 emit change events to the underlying mapping when the contents of
1005 the set are altered, including when values are added or removed.
1007 Note that :class:`.MutableSet` does **not** apply mutable tracking to the
1008 *values themselves* inside the set. Therefore it is not a sufficient
1009 solution for the use case of tracking deep changes to a *recursive*
1010 mutable structure. To support this use case,
1011 build a subclass of :class:`.MutableSet` that provides appropriate
1012 coercion to the values placed in the dictionary so that they too are
1013 "mutable", and emit events up to their parent structure.
1015 .. seealso::
1017 :class:`.MutableDict`
1019 :class:`.MutableList`
1022 """
1024 def update(self, *arg: Iterable[_T]) -> None:
1025 super().update(*arg)
1026 self.changed()
1028 def intersection_update(self, *arg: Iterable[Any]) -> None:
1029 super().intersection_update(*arg)
1030 self.changed()
1032 def difference_update(self, *arg: Iterable[Any]) -> None:
1033 super().difference_update(*arg)
1034 self.changed()
1036 def symmetric_difference_update(self, *arg: Iterable[_T]) -> None:
1037 super().symmetric_difference_update(*arg)
1038 self.changed()
1040 def __ior__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1041 self.update(other)
1042 return self
1044 def __iand__(self, other: AbstractSet[object]) -> MutableSet[_T]:
1045 self.intersection_update(other)
1046 return self
1048 def __ixor__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1049 self.symmetric_difference_update(other)
1050 return self
1052 def __isub__(self, other: AbstractSet[object]) -> MutableSet[_T]: # type: ignore[misc] # noqa: E501
1053 self.difference_update(other)
1054 return self
1056 def add(self, elem: _T) -> None:
1057 super().add(elem)
1058 self.changed()
1060 def remove(self, elem: _T) -> None:
1061 super().remove(elem)
1062 self.changed()
1064 def discard(self, elem: _T) -> None:
1065 super().discard(elem)
1066 self.changed()
1068 def pop(self, *arg: Any) -> _T:
1069 result = super().pop(*arg)
1070 self.changed()
1071 return result
1073 def clear(self) -> None:
1074 super().clear()
1075 self.changed()
1077 @classmethod
1078 def coerce(cls, index: str, value: Any) -> Optional[MutableSet[_T]]:
1079 """Convert plain set to instance of this class."""
1080 if not isinstance(value, cls):
1081 if isinstance(value, set):
1082 return cls(value)
1083 return Mutable.coerce(index, value)
1084 else:
1085 return value
1087 def __getstate__(self) -> Set[_T]:
1088 return set(self)
1090 def __setstate__(self, state: Iterable[_T]) -> None:
1091 self.update(state)
1093 def __reduce_ex__(
1094 self, proto: SupportsIndex
1095 ) -> Tuple[type, Tuple[List[int]]]:
1096 return (self.__class__, (list(self),))