codekingpro/portable-devtools
114k
1# ext/mutable.py
2# Copyright (C) 2005-2026 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
7
8r"""Provide support for tracking of in-place changes to scalar values,
9which are propagated into ORM change events on owning parent objects.
10
11.. _mutable_scalars:
12
13Establishing Mutability on Scalar Column Values
14===============================================
15
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::
20
21 from sqlalchemy.types import TypeDecorator, VARCHAR
22 import json
23
24
25 class JSONEncodedDict(TypeDecorator):
26 "Represents an immutable structure as a json-encoded string."
27
28 impl = VARCHAR
29
30 def process_bind_param(self, value, dialect):
31 if value is not None:
32 value = json.dumps(value)
33 return value
34
35 def process_result_value(self, value, dialect):
36 if value is not None:
37 value = json.loads(value)
38 return value
39
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.
44
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::
49
50 from sqlalchemy.ext.mutable import Mutable
51
52
53 class MutableDict(Mutable, dict):
54 @classmethod
55 def coerce(cls, key, value):
56 "Convert plain dictionaries to MutableDict."
57
58 if not isinstance(value, MutableDict):
59 if isinstance(value, dict):
60 return MutableDict(value)
61
62 # this call will raise ValueError
63 return Mutable.coerce(key, value)
64 else:
65 return value
66
67 def __setitem__(self, key, value):
68 "Detect dictionary set events and emit change events."
69
70 dict.__setitem__(self, key, value)
71 self.changed()
72
73 def __delitem__(self, key):
74 "Detect dictionary del events and emit change events."
75
76 dict.__delitem__(self, key)
77 self.changed()
78
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.
86
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``.
96
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::
103
104 from sqlalchemy import Table, Column, Integer
105
106 my_data = Table(
107 "my_data",
108 metadata,
109 Column("id", Integer, primary_key=True),
110 Column("data", MutableDict.as_mutable(JSONEncodedDict)),
111 )
112
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::
117
118 from sqlalchemy.orm import DeclarativeBase
119 from sqlalchemy.orm import Mapped
120 from sqlalchemy.orm import mapped_column
121
122
123 class Base(DeclarativeBase):
124 pass
125
126
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 )
133
134The ``MyDataClass.data`` member will now be notified of in place changes
135to its value.
136
137Any in-place changes to the ``MyDataClass.data`` member
138will flag the attribute as "dirty" on the parent object::
139
140 >>> from sqlalchemy.orm import Session
141
142 >>> sess = Session(some_engine)
143 >>> m1 = MyDataClass(data={"value1": "foo"})
144 >>> sess.add(m1)
145 >>> sess.commit()
146
147 >>> m1.data["value1"] = "bar"
148 >>> assert m1 in sess.dirty
149 True
150
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::
157
158 from sqlalchemy.orm import DeclarativeBase
159 from sqlalchemy.orm import Mapped
160 from sqlalchemy.orm import mapped_column
161
162 MutableDict.associate_with(JSONEncodedDict)
163
164
165 class Base(DeclarativeBase):
166 pass
167
168
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)
173
174Supporting Pickling
175--------------------
176
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::
188
189 class MyMutableType(Mutable):
190 def __getstate__(self):
191 d = self.__dict__.copy()
192 d.pop("_parents", None)
193 return d
194
195With our dictionary example, we need to return the contents of the dict itself
196(and also restore them on __setstate__)::
197
198 class MutableDict(Mutable, dict):
199 # ....
200
201 def __getstate__(self):
202 return dict(self)
203
204 def __setstate__(self, state):
205 self.update(state)
206
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.
211
212Receiving Events
213----------------
214
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::
219
220 from sqlalchemy.orm import DeclarativeBase
221 from sqlalchemy.orm import Mapped
222 from sqlalchemy.orm import mapped_column
223 from sqlalchemy import event
224
225
226 class Base(DeclarativeBase):
227 pass
228
229
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 )
236
237
238 @event.listens_for(MyDataClass.data, "modified")
239 def modified_json(instance, initiator):
240 print("json value modified:", instance.data)
241
242.. _mutable_composites:
243
244Establishing Mutability on Composites
245=====================================
246
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`.
251
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::
260
261 import dataclasses
262 from sqlalchemy.ext.mutable import MutableComposite
263
264
265 @dataclasses.dataclass
266 class Point(MutableComposite):
267 x: int
268 y: int
269
270 def __setattr__(self, key, value):
271 "Intercept set events"
272
273 # set the attribute
274 object.__setattr__(self, key, value)
275
276 # alert all parents to the change
277 self.changed()
278
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::
284
285 from sqlalchemy.orm import DeclarativeBase, Mapped
286 from sqlalchemy.orm import composite, mapped_column
287
288
289 class Base(DeclarativeBase):
290 pass
291
292
293 class Vertex(Base):
294 __tablename__ = "vertices"
295
296 id: Mapped[int] = mapped_column(primary_key=True)
297
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 )
304
305 def __repr__(self):
306 return f"Vertex(start={self.start}, end={self.end})"
307
308Any in-place changes to the ``Vertex.start`` or ``Vertex.end`` members
309will flag the attribute as "dirty" on the parent object:
310
311.. sourcecode:: python+sql
312
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)
321
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
329
330Coercing Mutable Composites
331---------------------------
332
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::
339
340 @dataclasses.dataclass
341 class Point(MutableComposite):
342 # other Point methods
343 # ...
344
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
351
352Supporting Pickling
353--------------------
354
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::
362
363 @dataclasses.dataclass
364 class Point(MutableComposite):
365 # ...
366
367 def __getstate__(self):
368 return self.x, self.y
369
370 def __setstate__(self, state):
371 self.x, self.y = state
372
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.
376
377""" # noqa: E501
378
379from __future__ import annotations
380
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
396
397from .. import event
398from .. import inspect
399from .. import types
400from ..orm import Mapper
401from ..orm._typing import _ExternalEntityType
402from ..orm._typing import _O
403from ..orm._typing import _T
404from ..orm.attributes import AttributeEventToken
405from ..orm.attributes import flag_modified
406from ..orm.attributes import InstrumentedAttribute
407from ..orm.attributes import QueryableAttribute
408from ..orm.context import QueryContext
409from ..orm.decl_api import DeclarativeAttributeIntercept
410from ..orm.state import InstanceState
411from ..orm.unitofwork import UOWTransaction
412from ..sql._typing import _TypeEngineArgument
413from ..sql.base import SchemaEventTarget
414from ..sql.schema import Column
415from ..sql.type_api import TypeEngine
416from ..util import memoized_property
417from ..util.typing import SupportsIndex
418
419_KT = TypeVar("_KT") # Key type.
420_VT = TypeVar("_VT") # Value type.
421
422
423class MutableBase:
424 """Common base class to :class:`.Mutable`
425 and :class:`.MutableComposite`.
426
427 """
428
429 @memoized_property
430 def _parents(self) -> WeakKeyDictionary[Any, Any]:
431 """Dictionary of parent object's :class:`.InstanceState`->attribute
432 name on the parent.
433
434 This attribute is a so-called "memoized" property. It initializes
435 itself with a new ``weakref.WeakKeyDictionary`` the first time
436 it is accessed, returning the same object upon subsequent access.
437
438 .. versionchanged:: 1.4 the :class:`.InstanceState` is now used
439 as the key in the weak dictionary rather than the instance
440 itself.
441
442 """
443
444 return weakref.WeakKeyDictionary()
445
446 @classmethod
447 def coerce(cls, key: str, value: Any) -> Optional[Any]:
448 """Given a value, coerce it into the target type.
449
450 Can be overridden by custom subclasses to coerce incoming
451 data into a particular type.
452
453 By default, raises ``ValueError``.
454
455 This method is called in different scenarios depending on if
456 the parent class is of type :class:`.Mutable` or of type
457 :class:`.MutableComposite`. In the case of the former, it is called
458 for both attribute-set operations as well as during ORM loading
459 operations. For the latter, it is only called during attribute-set
460 operations; the mechanics of the :func:`.composite` construct
461 handle coercion during load operations.
462
463
464 :param key: string name of the ORM-mapped attribute being set.
465 :param value: the incoming value.
466 :return: the method should return the coerced value, or raise
467 ``ValueError`` if the coercion cannot be completed.
468
469 """
470 if value is None:
471 return None
472 msg = "Attribute '%s' does not accept objects of type %s"
473 raise ValueError(msg % (key, type(value)))
474
475 @classmethod
476 def _get_listen_keys(cls, attribute: QueryableAttribute[Any]) -> Set[str]:
477 """Given a descriptor attribute, return a ``set()`` of the attribute
478 keys which indicate a change in the state of this attribute.
479
480 This is normally just ``set([attribute.key])``, but can be overridden
481 to provide for additional keys. E.g. a :class:`.MutableComposite`
482 augments this set with the attribute keys associated with the columns
483 that comprise the composite value.
484
485 This collection is consulted in the case of intercepting the
486 :meth:`.InstanceEvents.refresh` and
487 :meth:`.InstanceEvents.refresh_flush` events, which pass along a list
488 of attribute names that have been refreshed; the list is compared
489 against this set to determine if action needs to be taken.
490
491 """
492 return {attribute.key}
493
494 @classmethod
495 def _listen_on_attribute(
496 cls,
497 attribute: QueryableAttribute[Any],
498 coerce: bool,
499 parent_cls: _ExternalEntityType[Any],
500 ) -> None:
501 """Establish this type as a mutation listener for the given
502 mapped descriptor.
503
504 """
505 key = attribute.key
506 if parent_cls is not attribute.class_:
507 return
508
509 # rely on "propagate" here
510 parent_cls = attribute.class_
511
512 listen_keys = cls._get_listen_keys(attribute)
513
514 def load(state: InstanceState[_O], *args: Any) -> None:
515 """Listen for objects loaded or refreshed.
516
517 Wrap the target data member's value with
518 ``Mutable``.
519
520 """
521 val = state.dict.get(key, None)
522 if val is not None:
523 if coerce:
524 val = cls.coerce(key, val)
525 assert val is not None
526 state.dict[key] = val
527 val._parents[state] = key
528
529 def load_attrs(
530 state: InstanceState[_O],
531 ctx: Union[object, QueryContext, UOWTransaction],
532 attrs: Iterable[Any],
533 ) -> None:
534 if not attrs or listen_keys.intersection(attrs):
535 load(state)
536
537 def set_(
538 target: InstanceState[_O],
539 value: MutableBase | None,
540 oldvalue: MutableBase | None,
541 initiator: AttributeEventToken,
542 ) -> MutableBase | None:
543 """Listen for set/replace events on the target
544 data member.
545
546 Establish a weak reference to the parent object
547 on the incoming value, remove it for the one
548 outgoing.
549
550 """
551 if value is oldvalue:
552 return value
553
554 if not isinstance(value, cls):
555 value = cls.coerce(key, value)
556 if value is not None:
557 value._parents[target] = key
558 if isinstance(oldvalue, cls):
559 oldvalue._parents.pop(inspect(target), None)
560 return value
561
562 def pickle(
563 state: InstanceState[_O], state_dict: Dict[str, Any]
564 ) -> None:
565 val = state.dict.get(key, None)
566 if val is not None:
567 if "ext.mutable.values" not in state_dict:
568 state_dict["ext.mutable.values"] = defaultdict(list)
569 state_dict["ext.mutable.values"][key].append(val)
570
571 def unpickle(
572 state: InstanceState[_O], state_dict: Dict[str, Any]
573 ) -> None:
574 if "ext.mutable.values" in state_dict:
575 collection = state_dict["ext.mutable.values"]
576 if isinstance(collection, list):
577 # legacy format
578 for val in collection:
579 val._parents[state] = key
580 else:
581 for val in state_dict["ext.mutable.values"][key]:
582 val._parents[state] = key
583
584 event.listen(
585 parent_cls,
586 "_sa_event_merge_wo_load",
587 load,
588 raw=True,
589 propagate=True,
590 )
591
592 event.listen(parent_cls, "load", load, raw=True, propagate=True)
593 event.listen(
594 parent_cls, "refresh", load_attrs, raw=True, propagate=True
595 )
596 event.listen(
597 parent_cls, "refresh_flush", load_attrs, raw=True, propagate=True
598 )
599 event.listen(
600 attribute, "set", set_, raw=True, retval=True, propagate=True
601 )
602 event.listen(parent_cls, "pickle", pickle, raw=True, propagate=True)
603 event.listen(
604 parent_cls, "unpickle", unpickle, raw=True, propagate=True
605 )
606
607
608class Mutable(MutableBase):
609 """Mixin that defines transparent propagation of change
610 events to a parent object.
611
612 See the example in :ref:`mutable_scalars` for usage information.
613
614 """
615
616 def changed(self) -> None:
617 """Subclasses should call this method whenever change events occur."""
618
619 for parent, key in self._parents.items():
620 flag_modified(parent.obj(), key)
621
622 @classmethod
623 def associate_with_attribute(
624 cls, attribute: InstrumentedAttribute[_O]
625 ) -> None:
626 """Establish this type as a mutation listener for the given
627 mapped descriptor.
628
629 """
630 cls._listen_on_attribute(attribute, True, attribute.class_)
631
632 @classmethod
633 def associate_with(cls, sqltype: type) -> None:
634 """Associate this wrapper with all future mapped columns
635 of the given type.
636
637 This is a convenience method that calls
638 ``associate_with_attribute`` automatically.
639
640 .. warning::
641
642 The listeners established by this method are *global*
643 to all mappers, and are *not* garbage collected. Only use
644 :meth:`.associate_with` for types that are permanent to an
645 application, not with ad-hoc types else this will cause unbounded
646 growth in memory usage.
647
648 """
649
650 def listen_for_type(mapper: Mapper[_O], class_: type) -> None:
651 if mapper.non_primary:
652 return
653 for prop in mapper.column_attrs:
654 if isinstance(prop.columns[0].type, sqltype):
655 cls.associate_with_attribute(getattr(class_, prop.key))
656
657 event.listen(Mapper, "mapper_configured", listen_for_type)
658
659 @classmethod
660 def as_mutable(cls, sqltype: _TypeEngineArgument[_T]) -> TypeEngine[_T]:
661 """Associate a SQL type with this mutable Python type.
662
663 This establishes listeners that will detect ORM mappings against
664 the given type, adding mutation event trackers to those mappings.
665
666 The type is returned, unconditionally as an instance, so that
667 :meth:`.as_mutable` can be used inline::
668
669 Table(
670 "mytable",
671 metadata,
672 Column("id", Integer, primary_key=True),
673 Column("data", MyMutableType.as_mutable(PickleType)),
674 )
675
676 Note that the returned type is always an instance, even if a class
677 is given, and that only columns which are declared specifically with
678 that type instance receive additional instrumentation.
679
680 To associate a particular mutable type with all occurrences of a
681 particular type, use the :meth:`.Mutable.associate_with` classmethod
682 of the particular :class:`.Mutable` subclass to establish a global
683 association.
684
685 .. warning::
686
687 The listeners established by this method are *global*
688 to all mappers, and are *not* garbage collected. Only use
689 :meth:`.as_mutable` for types that are permanent to an application,
690 not with ad-hoc types else this will cause unbounded growth
691 in memory usage.
692
693 """
694 sqltype = types.to_instance(sqltype)
695
696 # a SchemaType will be copied when the Column is copied,
697 # and we'll lose our ability to link that type back to the original.
698 # so track our original type w/ columns
699 if isinstance(sqltype, SchemaEventTarget):
700
701 @event.listens_for(sqltype, "before_parent_attach")
702 def _add_column_memo(
703 sqltyp: TypeEngine[Any],
704 parent: Column[_T],
705 ) -> None:
706 parent.info["_ext_mutable_orig_type"] = sqltyp
707
708 schema_event_check = True
709 else:
710 schema_event_check = False
711
712 def listen_for_type(
713 mapper: Mapper[_T],
714 class_: Union[DeclarativeAttributeIntercept, type],
715 ) -> None:
716 if mapper.non_primary:
717 return
718 _APPLIED_KEY = "_ext_mutable_listener_applied"
719
720 for prop in mapper.column_attrs:
721 if (
722 # all Mutable types refer to a Column that's mapped,
723 # since this is the only kind of Core target the ORM can
724 # "mutate"
725 isinstance(prop.expression, Column)
726 and (
727 (
728 schema_event_check
729 and prop.expression.info.get(
730 "_ext_mutable_orig_type"
731 )
732 is sqltype
733 )
734 or prop.expression.type is sqltype
735 )
736 ):
737 if not prop.expression.info.get(_APPLIED_KEY, False):
738 prop.expression.info[_APPLIED_KEY] = True
739 cls.associate_with_attribute(getattr(class_, prop.key))
740
741 event.listen(Mapper, "mapper_configured", listen_for_type)
742
743 return sqltype
744
745
746class MutableComposite(MutableBase):
747 """Mixin that defines transparent propagation of change
748 events on a SQLAlchemy "composite" object to its
749 owning parent or parents.
750
751 See the example in :ref:`mutable_composites` for usage information.
752
753 """
754
755 @classmethod
756 def _get_listen_keys(cls, attribute: QueryableAttribute[_O]) -> Set[str]:
757 return {attribute.key}.union(attribute.property._attribute_keys)
758
759 def changed(self) -> None:
760 """Subclasses should call this method whenever change events occur."""
761
762 for parent, key in self._parents.items():
763 prop = parent.mapper.get_property(key)
764 for value, attr_name in zip(
765 prop._composite_values_from_instance(self),
766 prop._attribute_keys,
767 ):
768 setattr(parent.obj(), attr_name, value)
769
770
771def _setup_composite_listener() -> None:
772 def _listen_for_type(mapper: Mapper[_T], class_: type) -> None:
773 for prop in mapper.iterate_properties:
774 if (
775 hasattr(prop, "composite_class")
776 and isinstance(prop.composite_class, type)
777 and issubclass(prop.composite_class, MutableComposite)
778 ):
779 prop.composite_class._listen_on_attribute(
780 getattr(class_, prop.key), False, class_
781 )
782
783 if not event.contains(Mapper, "mapper_configured", _listen_for_type):
784 event.listen(Mapper, "mapper_configured", _listen_for_type)
785
786
787_setup_composite_listener()
788
789
790class MutableDict(Mutable, Dict[_KT, _VT]):
791 """A dictionary type that implements :class:`.Mutable`.
792
793 The :class:`.MutableDict` object implements a dictionary that will
794 emit change events to the underlying mapping when the contents of
795 the dictionary are altered, including when values are added or removed.
796
797 Note that :class:`.MutableDict` does **not** apply mutable tracking to the
798 *values themselves* inside the dictionary. Therefore it is not a sufficient
799 solution for the use case of tracking deep changes to a *recursive*
800 dictionary structure, such as a JSON structure. To support this use case,
801 build a subclass of :class:`.MutableDict` that provides appropriate
802 coercion to the values placed in the dictionary so that they too are
803 "mutable", and emit events up to their parent structure.
804
805 .. seealso::
806
807 :class:`.MutableList`
808
809 :class:`.MutableSet`
810
811 """
812
813 def __setitem__(self, key: _KT, value: _VT) -> None:
814 """Detect dictionary set events and emit change events."""
815 dict.__setitem__(self, key, value)
816 self.changed()
817
818 if TYPE_CHECKING:
819 # from https://github.com/python/mypy/issues/14858
820
821 @overload
822 def setdefault(
823 self: MutableDict[_KT, Optional[_T]], key: _KT, value: None = None
824 ) -> Optional[_T]: ...
825
826 @overload
827 def setdefault(self, key: _KT, value: _VT) -> _VT: ...
828
829 def setdefault(self, key: _KT, value: object = None) -> object: ...
830
831 else:
832
833 def setdefault(self, *arg): # noqa: F811
834 result = dict.setdefault(self, *arg)
835 self.changed()
836 return result
837
838 def __delitem__(self, key: _KT) -> None:
839 """Detect dictionary del events and emit change events."""
840 dict.__delitem__(self, key)
841 self.changed()
842
843 def update(self, *a: Any, **kw: _VT) -> None:
844 dict.update(self, *a, **kw)
845 self.changed()
846
847 if TYPE_CHECKING:
848
849 @overload
850 def pop(self, __key: _KT) -> _VT: ...
851
852 @overload
853 def pop(self, __key: _KT, __default: _VT | _T) -> _VT | _T: ...
854
855 def pop(
856 self, __key: _KT, __default: _VT | _T | None = None
857 ) -> _VT | _T: ...
858
859 else:
860
861 def pop(self, *arg): # noqa: F811
862 result = dict.pop(self, *arg)
863 self.changed()
864 return result
865
866 def popitem(self) -> Tuple[_KT, _VT]:
867 result = dict.popitem(self)
868 self.changed()
869 return result
870
871 def clear(self) -> None:
872 dict.clear(self)
873 self.changed()
874
875 @classmethod
876 def coerce(cls, key: str, value: Any) -> MutableDict[_KT, _VT] | None:
877 """Convert plain dictionary to instance of this class."""
878 if not isinstance(value, cls):
879 if isinstance(value, dict):
880 return cls(value)
881 return Mutable.coerce(key, value)
882 else:
883 return value
884
885 def __getstate__(self) -> Dict[_KT, _VT]:
886 return dict(self)
887
888 def __setstate__(
889 self, state: Union[Dict[str, int], Dict[str, str]]
890 ) -> None:
891 self.update(state)
892
893
894class MutableList(Mutable, List[_T]):
895 """A list type that implements :class:`.Mutable`.
896
897 The :class:`.MutableList` object implements a list that will
898 emit change events to the underlying mapping when the contents of
899 the list are altered, including when values are added or removed.
900
901 Note that :class:`.MutableList` does **not** apply mutable tracking to the
902 *values themselves* inside the list. Therefore it is not a sufficient
903 solution for the use case of tracking deep changes to a *recursive*
904 mutable structure, such as a JSON structure. To support this use case,
905 build a subclass of :class:`.MutableList` that provides appropriate
906 coercion to the values placed in the dictionary so that they too are
907 "mutable", and emit events up to their parent structure.
908
909 .. seealso::
910
911 :class:`.MutableDict`
912
913 :class:`.MutableSet`
914
915 """
916
917 def __reduce_ex__(
918 self, proto: SupportsIndex
919 ) -> Tuple[type, Tuple[List[int]]]:
920 return (self.__class__, (list(self),))
921
922 # needed for backwards compatibility with
923 # older pickles
924 def __setstate__(self, state: Iterable[_T]) -> None:
925 self[:] = state
926
927 def __setitem__(
928 self, index: SupportsIndex | slice, value: _T | Iterable[_T]
929 ) -> None:
930 """Detect list set events and emit change events."""
931 list.__setitem__(self, index, value)
932 self.changed()
933
934 def __delitem__(self, index: SupportsIndex | slice) -> None:
935 """Detect list del events and emit change events."""
936 list.__delitem__(self, index)
937 self.changed()
938
939 def pop(self, *arg: SupportsIndex) -> _T:
940 result = list.pop(self, *arg)
941 self.changed()
942 return result
943
944 def append(self, x: _T) -> None:
945 list.append(self, x)
946 self.changed()
947
948 def extend(self, x: Iterable[_T]) -> None:
949 list.extend(self, x)
950 self.changed()
951
952 def __iadd__(self, x: Iterable[_T]) -> MutableList[_T]: # type: ignore[override,misc] # noqa: E501
953 self.extend(x)
954 return self
955
956 def insert(self, i: SupportsIndex, x: _T) -> None:
957 list.insert(self, i, x)
958 self.changed()
959
960 def remove(self, i: _T) -> None:
961 list.remove(self, i)
962 self.changed()
963
964 def clear(self) -> None:
965 list.clear(self)
966 self.changed()
967
968 def sort(self, **kw: Any) -> None:
969 list.sort(self, **kw)
970 self.changed()
971
972 def reverse(self) -> None:
973 list.reverse(self)
974 self.changed()
975
976 @classmethod
977 def coerce(
978 cls, key: str, value: MutableList[_T] | _T
979 ) -> Optional[MutableList[_T]]:
980 """Convert plain list to instance of this class."""
981 if not isinstance(value, cls):
982 if isinstance(value, list):
983 return cls(value)
984 return Mutable.coerce(key, value)
985 else:
986 return value
987
988
989class MutableSet(Mutable, Set[_T]):
990 """A set type that implements :class:`.Mutable`.
991
992 The :class:`.MutableSet` object implements a set that will
993 emit change events to the underlying mapping when the contents of
994 the set are altered, including when values are added or removed.
995
996 Note that :class:`.MutableSet` does **not** apply mutable tracking to the
997 *values themselves* inside the set. Therefore it is not a sufficient
998 solution for the use case of tracking deep changes to a *recursive*
999 mutable structure. To support this use case,
1000 build a subclass of :class:`.MutableSet` that provides appropriate
1001 coercion to the values placed in the dictionary so that they too are
1002 "mutable", and emit events up to their parent structure.
1003
1004 .. seealso::
1005
1006 :class:`.MutableDict`
1007
1008 :class:`.MutableList`
1009
1010
1011 """
1012
1013 def update(self, *arg: Iterable[_T]) -> None:
1014 set.update(self, *arg)
1015 self.changed()
1016
1017 def intersection_update(self, *arg: Iterable[Any]) -> None:
1018 set.intersection_update(self, *arg)
1019 self.changed()
1020
1021 def difference_update(self, *arg: Iterable[Any]) -> None:
1022 set.difference_update(self, *arg)
1023 self.changed()
1024
1025 def symmetric_difference_update(self, *arg: Iterable[_T]) -> None:
1026 set.symmetric_difference_update(self, *arg)
1027 self.changed()
1028
1029 def __ior__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1030 self.update(other)
1031 return self
1032
1033 def __iand__(self, other: AbstractSet[object]) -> MutableSet[_T]:
1034 self.intersection_update(other)
1035 return self
1036
1037 def __ixor__(self, other: AbstractSet[_T]) -> MutableSet[_T]: # type: ignore[override,misc] # noqa: E501
1038 self.symmetric_difference_update(other)
1039 return self
1040
1041 def __isub__(self, other: AbstractSet[object]) -> MutableSet[_T]: # type: ignore[misc,unused-ignore] # noqa: E501
1042 self.difference_update(other)
1043 return self
1044
1045 def add(self, elem: _T) -> None:
1046 set.add(self, elem)
1047 self.changed()
1048
1049 def remove(self, elem: _T) -> None:
1050 set.remove(self, elem)
1051 self.changed()
1052
1053 def discard(self, elem: _T) -> None: # type: ignore[override,unused-ignore] # noqa: E501
1054 set.discard(self, elem)
1055 self.changed()
1056
1057 def pop(self, *arg: Any) -> _T:
1058 result = set.pop(self, *arg)
1059 self.changed()
1060 return result
1061
1062 def clear(self) -> None:
1063 set.clear(self)
1064 self.changed()
1065
1066 @classmethod
1067 def coerce(cls, index: str, value: Any) -> Optional[MutableSet[_T]]:
1068 """Convert plain set to instance of this class."""
1069 if not isinstance(value, cls):
1070 if isinstance(value, set):
1071 return cls(value)
1072 return Mutable.coerce(index, value)
1073 else:
1074 return value
1075
1076 def __getstate__(self) -> Set[_T]:
1077 return set(self)
1078
1079 def __setstate__(self, state: Iterable[_T]) -> None:
1080 self.update(state)
1081
1082 def __reduce_ex__(
1083 self, proto: SupportsIndex
1084 ) -> Tuple[type, Tuple[List[int]]]:
1085 return (self.__class__, (list(self),))
1086