From 6b1c08b33a8de8d3aab13d4111cd35cce034990b Mon Sep 17 00:00:00 2001 From: alisterburt Date: Mon, 2 May 2022 18:45:35 +0100 Subject: [PATCH] port `EventedDict` from napari (#79) * first pass at dict * remove key method * add interface parity test * mock events properly * first events tests * event tests done * black/flake * mypy progress * more mypy * even more mypy * the keys * 7 * 6 * 5 * 4 * 3 * 2 * is this the end * hold your breath and jsut ignore * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * add empty dict test * no cover for unused delitem of TypedMutableMapping interface * len method test * repr test * basetype tests * newlike test * copy test * fix pragma * update docs and types * minor updates Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Talley Lambert --- psygnal/containers/__init__.py | 2 + psygnal/containers/_evented_dict.py | 168 ++++++++++++++++++++++++++ tests/containers/test_evented_dict.py | 113 +++++++++++++++++ 3 files changed, 283 insertions(+) create mode 100644 psygnal/containers/_evented_dict.py create mode 100644 tests/containers/test_evented_dict.py diff --git a/psygnal/containers/__init__.py b/psygnal/containers/__init__.py index b65d39c6..08cfe556 100644 --- a/psygnal/containers/__init__.py +++ b/psygnal/containers/__init__.py @@ -1,12 +1,14 @@ """Containers backed by psygnal events.""" from typing import Any +from ._evented_dict import EventedDict from ._evented_list import EventedList from ._evented_set import EventedOrderedSet, EventedSet, OrderedSet from ._selectable_evented_list import SelectableEventedList from ._selection import Selection __all__ = [ + "EventedDict", "EventedList", "EventedObjectProxy", "EventedOrderedSet", diff --git a/psygnal/containers/_evented_dict.py b/psygnal/containers/_evented_dict.py new file mode 100644 index 00000000..e897dbbe --- /dev/null +++ b/psygnal/containers/_evented_dict.py @@ -0,0 +1,168 @@ +"""Dict that emits events when altered.""" + +from typing import ( + Dict, + Iterable, + Iterator, + Mapping, + MutableMapping, + Optional, + Sequence, + Tuple, + Type, + TypeVar, + Union, +) + +from .. import Signal, SignalGroup + +_K = TypeVar("_K") +_V = TypeVar("_V") +TypeOrSequenceOfTypes = Union[Type[_V], Sequence[Type[_V]]] +DictArg = Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]]] + + +class TypedMutableMapping(MutableMapping[_K, _V]): + """Dictionary that enforces value type. + + Parameters + ---------- + data : Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]], None], optional + Data suitable of passing to dict(). Mapping of {key: value} pairs, or + Iterable of two-tuples [(key, value), ...], or None to create an + basetype : TypeOrSequenceOfTypes, optional + Type or Sequence of Type objects. If provided, values entered into this Mapping + must be an instance of one of the provided types. by default () + """ + + def __init__( + self, + data: Optional[DictArg] = None, + *, + basetype: TypeOrSequenceOfTypes = (), + **kwargs: _V, + ): + + self._dict: Dict[_K, _V] = {} + self._basetypes: Tuple[Type[_V], ...] = ( + tuple(basetype) if isinstance(basetype, Sequence) else (basetype,) + ) + self.update({} if data is None else data, **kwargs) + + def __setitem__(self, key: _K, value: _V) -> None: + self._dict[key] = self._type_check(value) + + def __delitem__(self, key: _K) -> None: + del self._dict[key] + + def __getitem__(self, key: _K) -> _V: + return self._dict[key] + + def __len__(self) -> int: + return len(self._dict) + + def __iter__(self) -> Iterator[_K]: + return iter(self._dict) + + def __repr__(self) -> str: + return repr(self._dict) + + def _type_check(self, value: _V) -> _V: + """Check the types of items if basetypes are set for the model.""" + if self._basetypes and not any(isinstance(value, t) for t in self._basetypes): + raise TypeError( + f"Cannot add object with type {type(value)} to TypedDict expecting" + f"type {self._basetypes}" + ) + return value + + def __newlike__( + self, mapping: MutableMapping[_K, _V] + ) -> "TypedMutableMapping[_K, _V]": + new = self.__class__() + # separating this allows subclasses to omit these from their `__init__` + new._basetypes = self._basetypes + new.update(mapping) + return new + + def copy(self) -> "TypedMutableMapping[_K, _V]": + """Return a shallow copy of the dictionary.""" + return self.__newlike__(self) + + +class DictEvents(SignalGroup): + """Events available on EventedDict. + + Attributes + ---------- + adding (key: _K) + emitted before an item is added at `key` + added (key: _K, value: _V) + emitted after a `value` is added at `key` + changing (key: _K, old_value: _V, value: _V) + emitted before `old_value` is replaced with `value` at `key` + changed (key: _K, old_value: _V, value: _V) + emitted before `old_value` is replaced with `value` at `key` + removing (key: _K) + emitted before an item is removed at `key` + removed (key: _K, value: _V) + emitted after `value` is removed at `index` + """ + + adding = Signal(object) # (key, ) + added = Signal(object, object) # (key, value) + changing = Signal(object) # (key, ) + changed = Signal(object, object, object) # (key, old_value, value) + removing = Signal(object) # (key, ) + removed = Signal(object, object) # (key, value) + + +class EventedDict(TypedMutableMapping[_K, _V]): + """Mutable dictionary that emits events when altered. + + This class is designed to behave exactly like the builtin `dict`, but + will emit events before and after all mutations (addition, removal, and + changing). + + Parameters + ---------- + data : Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]], None], optional + Data suitable of passing to dict(). Mapping of {key: value} pairs, or + Iterable of two-tuples [(key, value), ...], or None to create an + basetype : TypeOrSequenceOfTypes, optional + Type or Sequence of Type objects. If provided, values entered into this Mapping + must be an instance of one of the provided types. by default (). + """ + + events: DictEvents # pragma: no cover + + def __init__( + self, + data: Optional[DictArg] = None, + *, + basetype: TypeOrSequenceOfTypes = (), + **kwargs: _V, + ): + self.events = DictEvents() + super().__init__(data, basetype=basetype, **kwargs) + + def __setitem__(self, key: _K, value: _V) -> None: + if key not in self._dict: + self.events.adding.emit(key) + super().__setitem__(key, value) + self.events.added.emit(key, value) + else: + old_value = self._dict[key] + if value is not old_value: + self.events.changing.emit(key) + super().__setitem__(key, value) + self.events.changed.emit(key, old_value, value) + + def __delitem__(self, key: _K) -> None: + item = self._dict[key] + self.events.removing.emit(key) + super().__delitem__(key) + self.events.removed.emit(key, item) + + def __repr__(self) -> str: + return f"EventedDict({super().__repr__()})" diff --git a/tests/containers/test_evented_dict.py b/tests/containers/test_evented_dict.py new file mode 100644 index 00000000..42e35a10 --- /dev/null +++ b/tests/containers/test_evented_dict.py @@ -0,0 +1,113 @@ +from unittest.mock import Mock + +import pytest + +from psygnal.containers._evented_dict import EventedDict + + +@pytest.fixture +def regular_dict(): + return {"A": 1, "B": 2, "C": 3} + + +@pytest.fixture +def test_dict(regular_dict): + """EventedDict without basetype set.""" + test_dict = EventedDict(regular_dict) + test_dict.events = Mock(wraps=test_dict.events) + return test_dict + + +@pytest.mark.parametrize( + "method_name, args, expected", + [ + ("__getitem__", ("A",), 1), # read + ("__setitem__", ("A", 3), None), # update + ("__setitem__", ("D", 3), None), # add new entry + ("__delitem__", ("A",), None), # delete + ("__len__", (), 3), + ("__newlike__", ({"A": 1},), {"A": 1}), + ("copy", (), {"A": 1, "B": 2, "C": 3}), + ], +) +def test_dict_interface_parity(regular_dict, test_dict, method_name, args, expected): + """Test that EventedDict interface is equivalent to the builtin dict.""" + test_dict_method = getattr(test_dict, method_name) + assert test_dict == regular_dict + if hasattr(regular_dict, method_name): + regular_dict_method = getattr(regular_dict, method_name) + assert test_dict_method(*args) == regular_dict_method(*args) == expected + assert test_dict == regular_dict + else: + test_dict_method(*args) # smoke test + + +def test_dict_inits(): + a = EventedDict({"A": 1, "B": 2, "C": 3}) + b = EventedDict(A=1, B=2, C=3) + c = EventedDict({"A": 1}, B=2, C=3) + assert a == b == c + + +def test_dict_repr(test_dict): + assert repr(test_dict) == "EventedDict({'A': 1, 'B': 2, 'C': 3})" + + +def test_instantiation_without_data(): + """Test that EventedDict can be instantiated without data.""" + test_dict = EventedDict() + assert isinstance(test_dict, EventedDict) + + +def test_basetype_enforcement_on_instantiation(): + """EventedDict with basetype set should enforce types on instantiation.""" + with pytest.raises(TypeError): + test_dict = EventedDict({"A": "not an int"}, basetype=int) + test_dict = EventedDict({"A": 1}) + assert isinstance(test_dict, EventedDict) + + +def test_basetype_enforcement_on_set_item(): + """EventedDict with basetype set should enforces types on setitem.""" + test_dict = EventedDict(basetype=int) + test_dict["A"] = 1 + with pytest.raises(TypeError): + test_dict["A"] = "not an int" + + +def test_dict_add_events(test_dict): + """Test that events are emitted before and after an item is added.""" + test_dict.events.adding.emit = Mock(wraps=test_dict.events.adding.emit) + test_dict.events.added.emit = Mock(wraps=test_dict.events.added.emit) + test_dict["D"] = 4 + test_dict.events.adding.emit.assert_called_with("D") + test_dict.events.added.emit.assert_called_with("D", 4) + + test_dict.events.adding.emit.reset_mock() + test_dict.events.added.emit.reset_mock() + test_dict["D"] = 4 + test_dict.events.adding.emit.assert_not_called() + test_dict.events.added.emit.assert_not_called() + + +def test_dict_change_events(test_dict): + """Test that events are emitted when an item in the dictionary is replaced.""" + # events shouldn't be emitted on addition + + test_dict.events.changing.emit = Mock(wraps=test_dict.events.changing.emit) + test_dict.events.changed.emit = Mock(wraps=test_dict.events.changed.emit) + test_dict["D"] = 4 + test_dict.events.changing.emit.assert_not_called() + test_dict.events.changed.emit.assert_not_called() + test_dict["C"] = 4 + test_dict.events.changing.emit.assert_called_with("C") + test_dict.events.changed.emit.assert_called_with("C", 3, 4) + + +def test_dict_remove_events(test_dict): + """Test that events are emitted before and after an item is removed.""" + test_dict.events.removing.emit = Mock(wraps=test_dict.events.removing.emit) + test_dict.events.removed.emit = Mock(wraps=test_dict.events.removed.emit) + test_dict.pop("C") + test_dict.events.removing.emit.assert_called_with("C") + test_dict.events.removed.emit.assert_called_with("C", 3)