Source code for

"""MutableMapping that emits events when altered."""
from typing import Mapping, Sequence, Type, Union

from import _K, _T, TypedMutableMapping
from import EmitterGroup, Event
from import SupportsEvents

[docs]class EventedDict(TypedMutableMapping[_K, _T]): """Mutable dictionary that emits events when altered. This class is designed to behave exactly like builtin ``dict``, but will emit events before and after all mutations (addition, removal, and changing). Parameters ---------- data : Mapping, optional Dictionary to initialize the class with. basetype : type of sequence of types, optional Type of the element in the dictionary. Events ------ changed (key: K, old_value: T, value: T) emitted when item at ``key`` is changed from ``old_value`` to ``value`` adding (key: K) emitted before an item is added to the dictionary with ``key`` added (key: K, value: T) emitted after ``value`` was added to the dictionary with ``key`` removing (key: K) emitted before ``key`` is removed from the dictionary removed (key: K, value: T) emitted after ``key`` was removed from the dictionary updated (key, K, value: T) emitted after ``value`` of ``key`` was changed. Only implemented by subclasses to give them an option to trigger some update after ``value`` was changed and this class did not register it. This can be useful if the ``basetype`` is not an evented object. """ events: EmitterGroup def __init__( self, data: Mapping[_K, _T] = None, basetype: Union[Type[_T], Sequence[Type[_T]]] = (), ) -> None: _events = { "changing": None, "changed": None, "adding": None, "added": None, "removing": None, "removed": None, "updated": None, } # For inheritance: If the mro already provides an EmitterGroup, add... if hasattr(self, "events") and isinstance(, EmitterGroup):**_events) else: # otherwise create a new one = EmitterGroup(source=self, **_events) super().__init__(data, basetype) def __setitem__(self, key: _K, value: _T): old = self._dict.get(key, None) if value is old or value == old: return if old is None: super().__setitem__(key, value), value=value) self._connect_child_emitters(value) else: super().__setitem__(key, value), old_value=old, value=value) def __delitem__(self, key: _K): self._disconnect_child_emitters(self[key]) item = self._dict.pop(key), value=item) def _reemit_child_event(self, event: Event): """An item in the dict emitted an event. Re-emit with key""" if not hasattr(event, "key"): event.key = self.key(event.source) # re-emit with this object's EventEmitter def _disconnect_child_emitters(self, child: _T): """Disconnect all events from the child from the re-emitter.""" if isinstance(child, SupportsEvents): def _connect_child_emitters(self, child: _T): """Connect all events from the child to be re-emitted.""" if isinstance(child, SupportsEvents): # make sure the event source has been set on the child if is None: = child
[docs] def key(self, value: _T): """Return first instance of value.""" for k, v in self._dict.items(): if v is value or v == value: return k return None