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 <[email protected]>

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",

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__()})"

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)