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-2024 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
24 class JSONEncodedDict(TypeDecorator):
25 "Represents an immutable structure as a json-encoded string."
27 impl = VARCHAR
29 def process_bind_param(self, value, dialect):
30 if value is not None:
31 value = json.dumps(value)
32 return value
34 def process_result_value(self, value, dialect):
35 if value is not None:
36 value = json.loads(value)
37 return value
39The usage of ``json`` is only for the purposes of example. The
40:mod:`sqlalchemy.ext.mutable` extension can be used
41with any type whose target Python type may be mutable, including
42:class:`.PickleType`, :class:`_postgresql.ARRAY`, etc.
44When using the :mod:`sqlalchemy.ext.mutable` extension, the value itself
45tracks all parents which reference it. Below, we illustrate a simple
46version of the :class:`.MutableDict` dictionary object, which applies
47the :class:`.Mutable` mixin to a plain Python dictionary::
49 from sqlalchemy.ext.mutable import Mutable
51 class MutableDict(Mutable, dict):
52 @classmethod
53 def coerce(cls, key, value):
54 "Convert plain dictionaries to MutableDict."
56 if not isinstance(value, MutableDict):
57 if isinstance(value, dict):
58 return MutableDict(value)
60 # this call will raise ValueError
61 return Mutable.coerce(key, value)
62 else:
63 return value
65 def __setitem__(self, key, value):
66 "Detect dictionary set events and emit change events."
68 dict.__setitem__(self, key, value)
69 self.changed()
71 def __delitem__(self, key):
72 "Detect dictionary del events and emit change events."
74 dict.__delitem__(self, key)
75 self.changed()
77The above dictionary class takes the approach of subclassing the Python
78built-in ``dict`` to produce a dict
79subclass which routes all mutation events through ``__setitem__``. There are
80variants on this approach, such as subclassing ``UserDict.UserDict`` or
81``collections.MutableMapping``; the part that's important to this example is
82that the :meth:`.Mutable.changed` method is called whenever an in-place
83change to the datastructure takes place.
85We also redefine the :meth:`.Mutable.coerce` method which will be used to
86convert any values that are not instances of ``MutableDict``, such
87as the plain dictionaries returned by the ``json`` module, into the
88appropriate type. Defining this method is optional; we could just as well
89created our ``JSONEncodedDict`` such that it always returns an instance
90of ``MutableDict``, and additionally ensured that all calling code
91uses ``MutableDict`` explicitly. When :meth:`.Mutable.coerce` is not
92overridden, any values applied to a parent object which are not instances
93of the mutable type will raise a ``ValueError``.
95Our new ``MutableDict`` type offers a class method
96:meth:`~.Mutable.as_mutable` which we can use within column metadata
97to associate with types. This method grabs the given type object or
98class and associates a listener that will detect all future mappings
99of this type, applying event listening instrumentation to the mapped
100attribute. Such as, with classical table metadata::
102 from sqlalchemy import Table, Column, Integer
104 my_data = Table('my_data', metadata,
105 Column('id', Integer, primary_key=True),
106 Column('data', MutableDict.as_mutable(JSONEncodedDict))
107 )
109Above, :meth:`~.Mutable.as_mutable` returns an instance of ``JSONEncodedDict``
110(if the type object was not an instance already), which will intercept any
111attributes which are mapped against this type. Below we establish a simple
112mapping against the ``my_data`` table::
114 from sqlalchemy.orm import DeclarativeBase
115 from sqlalchemy.orm import Mapped
116 from sqlalchemy.orm import mapped_column
118 class Base(DeclarativeBase):
119 pass
121 class MyDataClass(Base):
122 __tablename__ = 'my_data'
123 id: Mapped[int] = mapped_column(primary_key=True)
124 data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))
126The ``MyDataClass.data`` member will now be notified of in place changes
127to its value.
129Any in-place changes to the ``MyDataClass.data`` member
130will flag the attribute as "dirty" on the parent object::
132 >>> from sqlalchemy.orm import Session
134 >>> sess = Session(some_engine)
135 >>> m1 = MyDataClass(data={'value1':'foo'})
136 >>> sess.add(m1)
137 >>> sess.commit()
139 >>> m1.data['value1'] = 'bar'
140 >>> assert m1 in sess.dirty
141 True
143The ``MutableDict`` can be associated with all future instances
144of ``JSONEncodedDict`` in one step, using
145:meth:`~.Mutable.associate_with`. This is similar to
146:meth:`~.Mutable.as_mutable` except it will intercept all occurrences
147of ``MutableDict`` in all mappings unconditionally, without
148the need to declare it individually::
150 from sqlalchemy.orm import DeclarativeBase
151 from sqlalchemy.orm import Mapped
152 from sqlalchemy.orm import mapped_column
154 MutableDict.associate_with(JSONEncodedDict)
156 class Base(DeclarativeBase):
157 pass
159 class MyDataClass(Base):
160 __tablename__ = 'my_data'
161 id: Mapped[int] = mapped_column(primary_key=True)
162 data: Mapped[dict[str, str]] = mapped_column(JSONEncodedDict)
165Supporting Pickling
166--------------------
168The key to the :mod:`sqlalchemy.ext.mutable` extension relies upon the
169placement of a ``weakref.WeakKeyDictionary`` upon the value object, which
170stores a mapping of parent mapped objects keyed to the attribute name under
171which they are associated with this value. ``WeakKeyDictionary`` objects are
172not picklable, due to the fact that they contain weakrefs and function
173callbacks. In our case, this is a good thing, since if this dictionary were
174picklable, it could lead to an excessively large pickle size for our value
175objects that are pickled by themselves outside of the context of the parent.
176The developer responsibility here is only to provide a ``__getstate__`` method
177that excludes the :meth:`~MutableBase._parents` collection from the pickle
178stream::
180 class MyMutableType(Mutable):
181 def __getstate__(self):
182 d = self.__dict__.copy()
183 d.pop('_parents', None)
184 return d
186With our dictionary example, we need to return the contents of the dict itself
187(and also restore them on __setstate__)::
189 class MutableDict(Mutable, dict):
190 # ....
192 def __getstate__(self):
193 return dict(self)
195 def __setstate__(self, state):
196 self.update(state)
198In the case that our mutable value object is pickled as it is attached to one
199or more parent objects that are also part of the pickle, the :class:`.Mutable`
200mixin will re-establish the :attr:`.Mutable._parents` collection on each value
201object as the owning parents themselves are unpickled.
203Receiving Events
204----------------
206The :meth:`.AttributeEvents.modified` event handler may be used to receive
207an event when a mutable scalar emits a change event. This event handler
208is called when the :func:`.attributes.flag_modified` function is called
209from within the mutable extension::
211 from sqlalchemy.orm import DeclarativeBase
212 from sqlalchemy.orm import Mapped
213 from sqlalchemy.orm import mapped_column
214 from sqlalchemy import event
216 class Base(DeclarativeBase):
217 pass
219 class MyDataClass(Base):
220 __tablename__ = 'my_data'
221 id: Mapped[int] = mapped_column(primary_key=True)
222 data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))
224 @event.listens_for(MyDataClass.data, "modified")
225 def modified_json(instance, initiator):
226 print("json value modified:", instance.data)
228.. _mutable_composites:
230Establishing Mutability on Composites
231=====================================
233Composites are a special ORM feature which allow a single scalar attribute to
234be assigned an object value which represents information "composed" from one
235or more columns from the underlying mapped table. The usual example is that of
236a geometric "point", and is introduced in :ref:`mapper_composite`.
238As is the case with :class:`.Mutable`, the user-defined composite class
239subclasses :class:`.MutableComposite` as a mixin, and detects and delivers
240change events to its parents via the :meth:`.MutableComposite.changed` method.
241In the case of a composite class, the detection is usually via the usage of the
242special Python method ``__setattr__()``. In the example below, we expand upon the ``Point``
243class introduced in :ref:`mapper_composite` to include
244:class:`.MutableComposite` in its bases and to route attribute set events via
245``__setattr__`` to the :meth:`.MutableComposite.changed` method::
247 import dataclasses
248 from sqlalchemy.ext.mutable import MutableComposite
250 @dataclasses.dataclass
251 class Point(MutableComposite):
252 x: int
253 y: int
255 def __setattr__(self, key, value):
256 "Intercept set events"
258 # set the attribute
259 object.__setattr__(self, key, value)
261 # alert all parents to the change
262 self.changed()
265The :class:`.MutableComposite` class makes use of class mapping events to
266automatically establish listeners for any usage of :func:`_orm.composite` that
267specifies our ``Point`` type. Below, when ``Point`` is mapped to the ``Vertex``
268class, listeners are established which will route change events from ``Point``
269objects to each of the ``Vertex.start`` and ``Vertex.end`` attributes::
271 from sqlalchemy.orm import DeclarativeBase, Mapped
272 from sqlalchemy.orm import composite, mapped_column
274 class Base(DeclarativeBase):
275 pass
278 class Vertex(Base):
279 __tablename__ = "vertices"
281 id: Mapped[int] = mapped_column(primary_key=True)
283 start: Mapped[Point] = composite(mapped_column("x1"), mapped_column("y1"))
284 end: Mapped[Point] = composite(mapped_column("x2"), mapped_column("y2"))
286 def __repr__(self):
287 return f"Vertex(start={self.start}, end={self.end})"
289Any in-place changes to the ``Vertex.start`` or ``Vertex.end`` members
290will flag the attribute as "dirty" on the parent object:
292.. sourcecode:: python+sql
294 >>> from sqlalchemy.orm import Session
295 >>> sess = Session(engine)
296 >>> v1 = Vertex(start=Point(3, 4), end=Point(12, 15))
297 >>> sess.add(v1)
298 {sql}>>> sess.flush()
299 BEGIN (implicit)
300 INSERT INTO vertices (x1, y1, x2, y2) VALUES (?, ?, ?, ?)
301 [...] (3, 4, 12, 15)
303 {stop}>>> v1.end.x = 8
304 >>> assert v1 in sess.dirty
305 True
306 {sql}>>> sess.commit()
307 UPDATE vertices SET x2=? WHERE vertices.id = ?
308 [...] (8, 1)
309 COMMIT
311Coercing Mutable Composites
312---------------------------
314The :meth:`.MutableBase.coerce` method is also supported on composite types.
315In the case of :class:`.MutableComposite`, the :meth:`.MutableBase.coerce`
316method is only called for attribute set operations, not load operations.
317Overriding the :meth:`.MutableBase.coerce` method is essentially equivalent
318to using a :func:`.validates` validation routine for all attributes which
319make use of the custom composite type::
321 @dataclasses.dataclass
322 class Point(MutableComposite):
323 # other Point methods
324 # ...
326 def coerce(cls, key, value):
327 if isinstance(value, tuple):
328 value = Point(*value)
329 elif not isinstance(value, Point):
330 raise ValueError("tuple or Point expected")
331 return value
333Supporting Pickling
334--------------------
336As is the case with :class:`.Mutable`, the :class:`.MutableComposite` helper
337class uses a ``weakref.WeakKeyDictionary`` available via the
338:meth:`MutableBase._parents` attribute which isn't picklable. If we need to
339pickle instances of ``Point`` or its owning class ``Vertex``, we at least need
340to define a ``__getstate__`` that doesn't include the ``_parents`` dictionary.
341Below we define both a ``__getstate__`` and a ``__setstate__`` that package up
342the minimal form of our ``Point`` class::
344 @dataclasses.dataclass
345 class Point(MutableComposite):
346 # ...
348 def __getstate__(self):
349 return self.x, self.y
351 def __setstate__(self, state):
352 self.x, self.y = state
354As with :class:`.Mutable`, the :class:`.MutableComposite` augments the
355pickling process of the parent's object-relational state so that the
356:meth:`MutableBase._parents` collection is restored to all ``Point`` objects.
358""" # noqa: E501
360from __future__ import annotations
362from collections import defaultdict
363from typing import AbstractSet
364from typing import Any
365from typing import Dict
366from typing import Iterable
367from typing import List
368from typing import Optional
369from typing import overload
370from typing import Set
371from typing import SupportsIndex
372from typing import Tuple
373from typing import TYPE_CHECKING
374from typing import TypeVar
375from typing import Union
376import weakref
377from weakref import WeakKeyDictionary
379from .. import event
380from .. import inspect
381from .. import types
382from .. import util
383from ..orm import Mapper
384from ..orm._typing import _ExternalEntityType
385from ..orm._typing import _O
386from ..orm._typing import _T
387from ..orm.attributes import AttributeEventToken
388from ..orm.attributes import flag_modified
389from ..orm.attributes import InstrumentedAttribute
390from ..orm.attributes import QueryableAttribute
391from ..orm.context import QueryContext
392from ..orm.decl_api import DeclarativeAttributeIntercept
393from ..orm.state import InstanceState
394from ..orm.unitofwork import UOWTransaction
395from ..sql.base import SchemaEventTarget
396from ..sql.schema import Column
397from ..sql.type_api import TypeEngine
398from ..util import memoized_property
399from ..util.typing import TypeGuard
401_KT = TypeVar("_KT") # Key type.
402_VT = TypeVar("_VT") # Value type.
405class MutableBase:
406 """Common base class to :class:`.Mutable`
407 and :class:`.MutableComposite`.
409 """
411 @memoized_property
412 def _parents(self) -> WeakKeyDictionary[Any, Any]:
413 """Dictionary of parent object's :class:`.InstanceState`->attribute
414 name on the parent.
416 This attribute is a so-called "memoized" property. It initializes
417 itself with a new ``weakref.WeakKeyDictionary`` the first time
418 it is accessed, returning the same object upon subsequent access.
420 .. versionchanged:: 1.4 the :class:`.InstanceState` is now used
421 as the key in the weak dictionary rather than the instance
422 itself.
424 """
426 return weakref.WeakKeyDictionary()
428 @classmethod
429 def coerce(cls, key: str, value: Any) -> Optional[Any]:
430 """Given a value, coerce it into the target type.
432 Can be overridden by custom subclasses to coerce incoming
433 data into a particular type.
435 By default, raises ``ValueError``.
437 This method is called in different scenarios depending on if
438 the parent class is of type :class:`.Mutable` or of type
439 :class:`.MutableComposite`. In the case of the former, it is called
440 for both attribute-set operations as well as during ORM loading
441 operations. For the latter, it is only called during attribute-set
442 operations; the mechanics of the :func:`.composite` construct
443 handle coercion during load operations.
446 :param key: string name of the ORM-mapped attribute being set.
447 :param value: the incoming value.
448 :return: the method should return the coerced value, or raise
449 ``ValueError`` if the coercion cannot be completed.
451 """
452 if value is None:
453 return None
454 msg = "Attribute '%s' does not accept objects of type %s"
455 raise ValueError(msg % (key, type(value)))
457 @classmethod
458 def _get_listen_keys(cls, attribute: QueryableAttribute[Any]) -> Set[str]:
459 """Given a descriptor attribute, return a ``set()`` of the attribute
460 keys which indicate a change in the state of this attribute.
462 This is normally just ``set([attribute.key])``, but can be overridden
463 to provide for additional keys. E.g. a :class:`.MutableComposite`
464 augments this set with the attribute keys associated with the columns
465 that comprise the composite value.
467 This collection is consulted in the case of intercepting the
468 :meth:`.InstanceEvents.refresh` and
469 :meth:`.InstanceEvents.refresh_flush` events, which pass along a list
470 of attribute names that have been refreshed; the list is compared
471 against this set to determine if action needs to be taken.
473 """
474 return {attribute.key}
476 @classmethod
477 def _listen_on_attribute(
478 cls,
479 attribute: QueryableAttribute[Any],
480 coerce: bool,
481 parent_cls: _ExternalEntityType[Any],
482 ) -> None:
483 """Establish this type as a mutation listener for the given
484 mapped descriptor.
486 """
487 key = attribute.key
488 if parent_cls is not attribute.class_:
489 return
491 # rely on "propagate" here
492 parent_cls = attribute.class_
494 listen_keys = cls._get_listen_keys(attribute)
496 def load(state: InstanceState[_O], *args: Any) -> None:
497 """Listen for objects loaded or refreshed.
499 Wrap the target data member's value with
500 ``Mutable``.
502 """
503 val = state.dict.get(key, None)
504 if val is not None:
505 if coerce:
506 val = cls.coerce(key, val)
507 state.dict[key] = val
508 val._parents[state] = key
510 def load_attrs(
511 state: InstanceState[_O],
512 ctx: Union[object, QueryContext, UOWTransaction],
513 attrs: Iterable[Any],
514 ) -> None:
515 if not attrs or listen_keys.intersection(attrs):
516 load(state)
518 def set_(
519 target: InstanceState[_O],
520 value: MutableBase | None,
521 oldvalue: MutableBase | None,
522 initiator: AttributeEventToken,
523 ) -> MutableBase | None:
524 """Listen for set/replace events on the target
525 data member.
527 Establish a weak reference to the parent object
528 on the incoming value, remove it for the one
529 outgoing.
531 """
532 if value is oldvalue:
533 return value
535 if not isinstance(value, cls):
536 value = cls.coerce(key, value)
537 if value is not None:
538 value._parents[target] = key
539 if isinstance(oldvalue, cls):
540 oldvalue._parents.pop(inspect(target), None)
541 return value
543 def pickle(
544 state: InstanceState[_O], state_dict: Dict[str, Any]
545 ) -> None:
546 val = state.dict.get(key, None)
547 if val is not None:
548 if "ext.mutable.values" not in state_dict:
549 state_dict["ext.mutable.values"] = defaultdict(list)
550 state_dict["ext.mutable.values"][key].append(val)
552 def unpickle(
553 state: InstanceState[_O], state_dict: Dict[str, Any]
554 ) -> None:
555 if "ext.mutable.values" in state_dict:
556 collection = state_dict["ext.mutable.values"]
557 if isinstance(collection, list):
558 # legacy format
559 for val in collection:
560 val._parents[state] = key
561 else:
562 for val in state_dict["ext.mutable.values"][key]:
563 val._parents[state] = key
565 event.listen(
566 parent_cls,
567 "_sa_event_merge_wo_load",
568 load,
569 raw=True,
570 propagate=True,
571 )
573 event.listen(parent_cls, "load", load, raw=True, propagate=True)
574 event.listen(
575 parent_cls, "refresh", load_attrs, raw=True, propagate=True
576 )
577 event.listen(
578 parent_cls, "refresh_flush", load_attrs, raw=True, propagate=True
579 )
580 event.listen(
581 attribute, "set", set_, raw=True, retval=True, propagate=True
582 )
583 event.listen(parent_cls, "pickle", pickle, raw=True, propagate=True)
584 event.listen(
585 parent_cls, "unpickle", unpickle, raw=True, propagate=True
586 )
589class Mutable(MutableBase):
590 """Mixin that defines transparent propagation of change
591 events to a parent object.
593 See the example in :ref:`mutable_scalars` for usage information.
595 """
597 def changed(self) -> None:
598 """Subclasses should call this method whenever change events occur."""
600 for parent, key in self._parents.items():
601 flag_modified(parent.obj(), key)
603 @classmethod
604 def associate_with_attribute(
605 cls, attribute: InstrumentedAttribute[_O]
606 ) -> None:
607 """Establish this type as a mutation listener for the given
608 mapped descriptor.
610 """
611 cls._listen_on_attribute(attribute, True, attribute.class_)
613 @classmethod
614 def associate_with(cls, sqltype: type) -> None:
615 """Associate this wrapper with all future mapped columns
616 of the given type.
618 This is a convenience method that calls
619 ``associate_with_attribute`` automatically.
621 .. warning::
623 The listeners established by this method are *global*
624 to all mappers, and are *not* garbage collected. Only use
625 :meth:`.associate_with` for types that are permanent to an
626 application, not with ad-hoc types else this will cause unbounded
627 growth in memory usage.
629 """
631 def listen_for_type(mapper: Mapper[_O], class_: type) -> None:
632 if mapper.non_primary:
633 return
634 for prop in mapper.column_attrs:
635 if isinstance(prop.columns[0].type, sqltype):
636 cls.associate_with_attribute(getattr(class_, prop.key))
638 event.listen(Mapper, "mapper_configured", listen_for_type)
640 @classmethod
641 def as_mutable(cls, sqltype: TypeEngine[_T]) -> TypeEngine[_T]:
642 """Associate a SQL type with this mutable Python type.
644 This establishes listeners that will detect ORM mappings against
645 the given type, adding mutation event trackers to those mappings.
647 The type is returned, unconditionally as an instance, so that
648 :meth:`.as_mutable` can be used inline::
650 Table('mytable', metadata,
651 Column('id', Integer, primary_key=True),
652 Column('data', MyMutableType.as_mutable(PickleType))
653 )
655 Note that the returned type is always an instance, even if a class
656 is given, and that only columns which are declared specifically with
657 that type instance receive additional instrumentation.
659 To associate a particular mutable type with all occurrences of a
660 particular type, use the :meth:`.Mutable.associate_with` classmethod
661 of the particular :class:`.Mutable` subclass to establish a global
662 association.
664 .. warning::
666 The listeners established by this method are *global*
667 to all mappers, and are *not* garbage collected. Only use
668 :meth:`.as_mutable` for types that are permanent to an application,
669 not with ad-hoc types else this will cause unbounded growth
670 in memory usage.
672 """
673 sqltype = types.to_instance(sqltype)
675 # a SchemaType will be copied when the Column is copied,
676 # and we'll lose our ability to link that type back to the original.
677 # so track our original type w/ columns
678 if isinstance(sqltype, SchemaEventTarget):
680 @event.listens_for(sqltype, "before_parent_attach")
681 def _add_column_memo(
682 sqltyp: TypeEngine[Any],
683 parent: Column[_T],
684 ) -> None:
685 parent.info["_ext_mutable_orig_type"] = sqltyp
687 schema_event_check = True
688 else:
689 schema_event_check = False
691 def listen_for_type(
692 mapper: Mapper[_T],
693 class_: Union[DeclarativeAttributeIntercept, type],
694 ) -> None:
695 if mapper.non_primary:
696 return
697 _APPLIED_KEY = "_ext_mutable_listener_applied"
699 for prop in mapper.column_attrs:
700 if (
701 # all Mutable types refer to a Column that's mapped,
702 # since this is the only kind of Core target the ORM can
703 # "mutate"
704 isinstance(prop.expression, Column)
705 and (
706 (
707 schema_event_check
708 and prop.expression.info.get(
709 "_ext_mutable_orig_type"
710 )
711 is sqltype
712 )
713 or prop.expression.type is sqltype
714 )
715 ):
716 if not prop.expression.info.get(_APPLIED_KEY, False):
717 prop.expression.info[_APPLIED_KEY] = True
718 cls.associate_with_attribute(getattr(class_, prop.key))
720 event.listen(Mapper, "mapper_configured", listen_for_type)
722 return sqltype
725class MutableComposite(MutableBase):
726 """Mixin that defines transparent propagation of change
727 events on a SQLAlchemy "composite" object to its
728 owning parent or parents.
730 See the example in :ref:`mutable_composites` for usage information.
732 """
734 @classmethod
735 def _get_listen_keys(cls, attribute: QueryableAttribute[_O]) -> Set[str]:
736 return {attribute.key}.union(attribute.property._attribute_keys)
738 def changed(self) -> None:
739 """Subclasses should call this method whenever change events occur."""
741 for parent, key in self._parents.items():
742 prop = parent.mapper.get_property(key)
743 for value, attr_name in zip(
744 prop._composite_values_from_instance(self),
745 prop._attribute_keys,
746 ):
747 setattr(parent.obj(), attr_name, value)
750def _setup_composite_listener() -> None:
751 def _listen_for_type(mapper: Mapper[_T], class_: type) -> None:
752 for prop in mapper.iterate_properties:
753 if (
754 hasattr(prop, "composite_class")
755 and isinstance(prop.composite_class, type)
756 and issubclass(prop.composite_class, MutableComposite)
757 ):
758 prop.composite_class._listen_on_attribute(
759 getattr(class_, prop.key), False, class_
760 )
762 if not event.contains(Mapper, "mapper_configured", _listen_for_type):
763 event.listen(Mapper, "mapper_configured", _listen_for_type)
766_setup_composite_listener()
769class MutableDict(Mutable, Dict[_KT, _VT]):
770 """A dictionary type that implements :class:`.Mutable`.
772 The :class:`.MutableDict` object implements a dictionary that will
773 emit change events to the underlying mapping when the contents of
774 the dictionary are altered, including when values are added or removed.
776 Note that :class:`.MutableDict` does **not** apply mutable tracking to the
777 *values themselves* inside the dictionary. Therefore it is not a sufficient
778 solution for the use case of tracking deep changes to a *recursive*
779 dictionary structure, such as a JSON structure. To support this use case,
780 build a subclass of :class:`.MutableDict` that provides appropriate
781 coercion to the values placed in the dictionary so that they too are
782 "mutable", and emit events up to their parent structure.
784 .. seealso::
786 :class:`.MutableList`
788 :class:`.MutableSet`
790 """
792 def __setitem__(self, key: _KT, value: _VT) -> None:
793 """Detect dictionary set events and emit change events."""
794 super().__setitem__(key, value)
795 self.changed()
797 if TYPE_CHECKING:
798 # from https://github.com/python/mypy/issues/14858
800 @overload
801 def setdefault(
802 self: MutableDict[_KT, Optional[_T]], key: _KT, value: None = None
803 ) -> Optional[_T]: ...
805 @overload
806 def setdefault(self, key: _KT, value: _VT) -> _VT: ...
808 def setdefault(self, key: _KT, value: object = None) -> object: ...
810 else:
812 def setdefault(self, *arg): # noqa: F811
813 result = super().setdefault(*arg)
814 self.changed()
815 return result
817 def __delitem__(self, key: _KT) -> None:
818 """Detect dictionary del events and emit change events."""
819 super().__delitem__(key)
820 self.changed()
822 def update(self, *a: Any, **kw: _VT) -> None:
823 super().update(*a, **kw)
824 self.changed()
826 if TYPE_CHECKING:
828 @overload
829 def pop(self, __key: _KT, /) -> _VT: ...
831 @overload
832 def pop(self, __key: _KT, default: _VT | _T, /) -> _VT | _T: ...
834 def pop(
835 self, __key: _KT, __default: _VT | _T | None = None, /
836 ) -> _VT | _T: ...
838 else:
840 def pop(self, *arg): # noqa: F811
841 result = super().pop(*arg)
842 self.changed()
843 return result
845 def popitem(self) -> Tuple[_KT, _VT]:
846 result = super().popitem()
847 self.changed()
848 return result
850 def clear(self) -> None:
851 super().clear()
852 self.changed()
854 @classmethod
855 def coerce(cls, key: str, value: Any) -> MutableDict[_KT, _VT] | None:
856 """Convert plain dictionary to instance of this class."""
857 if not isinstance(value, cls):
858 if isinstance(value, dict):
859 return cls(value)
860 return Mutable.coerce(key, value)
861 else:
862 return value
864 def __getstate__(self) -> Dict[_KT, _VT]:
865 return dict(self)
867 def __setstate__(
868 self, state: Union[Dict[str, int], Dict[str, str]]
869 ) -> None:
870 self.update(state)
873class MutableList(Mutable, List[_T]):
874 """A list type that implements :class:`.Mutable`.
876 The :class:`.MutableList` object implements a list that will
877 emit change events to the underlying mapping when the contents of
878 the list are altered, including when values are added or removed.
880 Note that :class:`.MutableList` does **not** apply mutable tracking to the
881 *values themselves* inside the list. Therefore it is not a sufficient
882 solution for the use case of tracking deep changes to a *recursive*
883 mutable structure, such as a JSON structure. To support this use case,
884 build a subclass of :class:`.MutableList` that provides appropriate
885 coercion to the values placed in the dictionary so that they too are
886 "mutable", and emit events up to their parent structure.
888 .. seealso::
890 :class:`.MutableDict`
892 :class:`.MutableSet`
894 """
896 def __reduce_ex__(
897 self, proto: SupportsIndex
898 ) -> Tuple[type, Tuple[List[int]]]:
899 return (self.__class__, (list(self),))
901 # needed for backwards compatibility with
902 # older pickles
903 def __setstate__(self, state: Iterable[_T]) -> None:
904 self[:] = state
906 def is_scalar(self, value: _T | Iterable[_T]) -> TypeGuard[_T]:
907 return not util.is_non_string_iterable(value)
909 def is_iterable(self, value: _T | Iterable[_T]) -> TypeGuard[Iterable[_T]]:
910 return util.is_non_string_iterable(value)
912 def __setitem__(
913 self, index: SupportsIndex | slice, value: _T | Iterable[_T]
914 ) -> None:
915 """Detect list set events and emit change events."""
916 if isinstance(index, SupportsIndex) and self.is_scalar(value):
917 super().__setitem__(index, value)
918 elif isinstance(index, slice) and self.is_iterable(value):
919 super().__setitem__(index, value)
920 self.changed()
922 def __delitem__(self, index: SupportsIndex | slice) -> None:
923 """Detect list del events and emit change events."""
924 super().__delitem__(index)
925 self.changed()
927 def pop(self, *arg: SupportsIndex) -> _T:
928 result = super().pop(*arg)
929 self.changed()
930 return result
932 def append(self, x: _T) -> None:
933 super().append(x)
934 self.changed()
936 def extend(self, x: Iterable[_T]) -> None:
937 super().extend(x)
938 self.changed()
940 def __iadd__(self, x: Iterable[_T]) -> MutableList[_T]: # type: ignore[override,misc] # noqa: E501
941 self.extend(x)
942 return self
944 def insert(self, i: SupportsIndex, x: _T) -> None:
945 super().insert(i, x)
946 self.changed()
948 def remove(self, i: _T) -> None:
949 super().remove(i)
950 self.changed()
952 def clear(self) -> None:
953 super().clear()
954 self.changed()
956 def sort(self, **kw: Any) -> None:
957 super().sort(**kw)
958 self.changed()
960 def reverse(self) -> None:
961 super().reverse()
962 self.changed()
964 @classmethod
965 def coerce(
966 cls, key: str, value: MutableList[_T] | _T
967 ) -> Optional[MutableList[_T]]:
968 """Convert plain list to instance of this class."""
969 if not isinstance(value, cls):
970 if isinstance(value, list):
971 return cls(value)
972 return Mutable.coerce(key, value)
973 else:
974 return value
977class MutableSet(Mutable, Set[_T]):
978 """A set type that implements :class:`.Mutable`.
980 The :class:`.MutableSet` object implements a set that will
981 emit change events to the underlying mapping when the contents of
982 the set are altered, including when values are added or removed.
984 Note that :class:`.MutableSet` does **not** apply mutable tracking to the
985 *values themselves* inside the set. Therefore it is not a sufficient
986 solution for the use case of tracking deep changes to a *recursive*
987 mutable structure. To support this use case,
988 build a subclass of :class:`.MutableSet` that provides appropriate
989 coercion to the values placed in the dictionary so that they too are
990 "mutable", and emit events up to their parent structure.
992 .. seealso::
994 :class:`.MutableDict`
996 :class:`.MutableList`
999 """
1001 def update(self, *arg: Iterable[_T]) -> None:
1002 super().update(*arg)
1003 self.changed()
1005 def intersection_update(self, *arg: Iterable[Any]) -> None:
1006 super().intersection_update(*arg)
1007 self.changed()
1009 def difference_update(self, *arg: Iterable[Any]) -> None:
1010 super().difference_update(*arg)
1011 self.changed()
1013 def symmetric_difference_update(self, *arg: Iterable[_T]) -> None:
1014 super().symmetric_difference_update(*arg)
1015 self.changed()
1017 def __ior__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1018 self.update(other)
1019 return self
1021 def __iand__(self, other: AbstractSet[object]) -> MutableSet[_T]:
1022 self.intersection_update(other)
1023 return self
1025 def __ixor__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1026 self.symmetric_difference_update(other)
1027 return self
1029 def __isub__(self, other: AbstractSet[object]) -> MutableSet[_T]: # type: ignore[misc] # noqa: E501
1030 self.difference_update(other)
1031 return self
1033 def add(self, elem: _T) -> None:
1034 super().add(elem)
1035 self.changed()
1037 def remove(self, elem: _T) -> None:
1038 super().remove(elem)
1039 self.changed()
1041 def discard(self, elem: _T) -> None:
1042 super().discard(elem)
1043 self.changed()
1045 def pop(self, *arg: Any) -> _T:
1046 result = super().pop(*arg)
1047 self.changed()
1048 return result
1050 def clear(self) -> None:
1051 super().clear()
1052 self.changed()
1054 @classmethod
1055 def coerce(cls, index: str, value: Any) -> Optional[MutableSet[_T]]:
1056 """Convert plain set to instance of this class."""
1057 if not isinstance(value, cls):
1058 if isinstance(value, set):
1059 return cls(value)
1060 return Mutable.coerce(index, value)
1061 else:
1062 return value
1064 def __getstate__(self) -> Set[_T]:
1065 return set(self)
1067 def __setstate__(self, state: Iterable[_T]) -> None:
1068 self.update(state)
1070 def __reduce_ex__(
1071 self, proto: SupportsIndex
1072 ) -> Tuple[type, Tuple[List[int]]]:
1073 return (self.__class__, (list(self),))