Implement SetItemObserver for observing mutations on a set

docs/source/traits_api_reference/traits.observers.rst

@@ -15,6 +15,10 @@
     :undoc-members:
     :show-inheritance:
 
+.. autoclass:: SetChangeEvent
+   :members:
+   :inherited-members:
+
 .. autoclass:: TraitChangeEvent
    :members:
    :inherited-members:

traits/observers/_set_change_event.py

@@ -0,0 +1,61 @@
+# (C) Copyright 2005-2020 Enthought, Inc., Austin, TX
+# All rights reserved.
+#
+# This software is provided without warranty under the terms of the BSD
+# license included in LICENSE.txt and may be redistributed only under
+# the conditions described in the aforementioned license. The license
+# is also available online at http://www.enthought.com/licenses/BSD.txt
+#
+# Thanks for using Enthought open source!
+
+
+# SetChangeEvent is in the public API.
+
+
+class SetChangeEvent:
+    """ Event object to represent mutations on a set.
+
+    Attributes
+    ----------
+    trait_set : traits.trait_set_object.TraitSet
+        The set being mutated.
+    removed : set
+        Values removed from the set.
+    added : set
+        Values added to the set.
+    """
+
+    def __init__(self, *, trait_set, removed, added):
+        self.trait_set = trait_set
+        self.removed = removed
+        self.added = added
+
+    def __repr__(self):
+        return (
+            "{event.__class__.__name__}("
+            "trait_set={event.trait_set!r}, "
+            "removed={event.removed!r}, "
+            "added={event.added!r}"
+            ")".format(event=self)
+        )
+
+
+def set_event_factory(trait_set, removed, added):
+    """ Adapt the call signature of TraitSet.notify to create an event.
+
+    Parameters
+    ----------
+    trait_set : traits.trait_set_object.TraitSet
+        The set being mutated.
+    removed : set
+        Values removed from the set.
+    added : set
+        Values added to the set.
+
+    Returns
+    -------
+    SetChangeEvent
+    """
+    return SetChangeEvent(
+        trait_set=trait_set, added=added, removed=removed,
+    )

traits/observers/_set_item_observer.py

@@ -0,0 +1,203 @@
+# (C) Copyright 2005-2020 Enthought, Inc., Austin, TX
+# All rights reserved.
+#
+# This software is provided without warranty under the terms of the BSD
+# license included in LICENSE.txt and may be redistributed only under
+# the conditions described in the aforementioned license. The license
+# is also available online at http://www.enthought.com/licenses/BSD.txt
+#
+# Thanks for using Enthought open source!
+
+from traits.observers._i_observer import IObserver
+from traits.observers._observe import add_or_remove_notifiers
+from traits.observers._observer_change_notifier import ObserverChangeNotifier
+from traits.observers._set_change_event import set_event_factory
+from traits.observers._trait_event_notifier import TraitEventNotifier
+from traits.trait_set_object import TraitSet
+
+
+@IObserver.register
+class SetItemObserver:
+    """ Observer for observing mutations on a set.
+
+    Parameters
+    ----------
+    notify : boolean
+        Whether to notify for changes.
+    optional : boolean
+        If False, this observer will complain if the incoming object is not
+        an observable set. If True and the incoming object is not a set, this
+        observer will do nothing. Useful for the 'items' keyword in the text
+        parser, where the source container type is ambiguous.
+    """
+
+    def __init__(self, *, notify, optional):
+        self.notify = notify
+        self.optional = optional
+
+    def __hash__(self):
+        """ Return a hash of this object."""
+        return hash((type(self).__name__, self.notify, self.optional))
+
+    def __eq__(self, other):
+        """ Return true if this observer is equal to the given one."""
+        return (
+            type(self) is type(other)
+            and self.notify == other.notify
+            and self.optional == other.optional
+        )
+
+    def iter_observables(self, object):
+        """ If the given object is an observable set, yield that set.
+        Otherwise, raise an error, unless this observer is optional
+
+        Parameters
+        ----------
+        object: object
+            Object provided by another observers or by the user.
+
+        Yields
+        ------
+        IObservable
+
+        Raises
+        ------
+        ValueError
+            If the given object is not an observable set and optional is false.
+        """
+        if not isinstance(object, TraitSet):
+            if self.optional:
+                return
+            raise ValueError(
+                "Expected a TraitSet to be observed, "
+                "got {!r} (type: {!r})".format(object, type(object))
+            )
+
+        yield object
+
+    def iter_objects(self, object):
+        """ Yield the content of the set if the given object is an observable
+        set. Otherwise, raise an error, unless the observer is optional.
+
+        The content of the set will be passed onto the children observer(s)
+        following this one in an ObserverGraph.
+
+        Parameters
+        ----------
+        object: object
+            Object provided by another observers or by the user.
+
+        Yields
+        ------
+        value : object
+
+        Raises
+        ------
+        ValueError
+            If the given object is not an observable set and optional is false.
+        """
+        if not isinstance(object, TraitSet):
+            if self.optional:
+                return
+            raise ValueError(
+                "Expected a TraitSet to be observed, "
+                "got {!r} (type: {!r})".format(object, type(object))
+            )
+
+        yield from object
+
+    def get_notifier(self, handler, target, dispatcher):
+        """ Return a notifier for calling the user handler with the change
+        event.
+
+        Returns
+        -------
+        notifier : TraitEventNotifier
+        """
+        return TraitEventNotifier(
+            handler=handler,
+            target=target,
+            dispatcher=dispatcher,
+            event_factory=set_event_factory,
+            prevent_event=lambda event: False,
+        )
+
+    def get_maintainer(self, graph, handler, target, dispatcher):
+        """ Return a notifier for maintaining downstream observers when
+        a set is mutated.
+
+        Parameters
+        ----------
+        graph : ObserverGraph
+            Description for the *downstream* observers, i.e. excluding self.
+        handler : callable
+            User handler.
+        target : object
+            Object seen by the user as the owner of the observer.
+        dispatcher : callable
+            Callable for dispatching the handler.
+
+        Returns
+        -------
+        notifier : ObserverChangeNotifier
+        """
+        return ObserverChangeNotifier(
+            observer_handler=_observer_change_handler,
+            event_factory=set_event_factory,
+            prevent_event=lambda event: False,
+            graph=graph,
+            handler=handler,
+            target=target,
+            dispatcher=dispatcher,
+        )
+
+    def iter_extra_graphs(self, graph):
+        """ Yield new ObserverGraph to be contributed by this observer.
+
+        Parameters
+        ----------
+        graph : ObserverGraph
+            The graph this observer is part of.
+
+        Yields
+        ------
+        ObserverGraph
+        """
+        # Unlike CTrait, no need to handle trait_added
+        yield from ()
+
+
+def _observer_change_handler(event, graph, handler, target, dispatcher):
+    """ Handler for maintaining observers. Used by ObserverChangeNotifier.
+
+    Parameters
+    ----------
+    event : SetChangeEvent
+        Change event that triggers the maintainer.
+    graph : ObserverGraph
+        Description for the *downstream* observers, i.e. excluding self.
+    handler : callable
+        User handler.
+    target : object
+        Object seen by the user as the owner of the observer.
+    dispatcher : callable
+        Callable for dispatching the handler.
+    """
+    for removed_item in event.removed:
+        add_or_remove_notifiers(
+            object=removed_item,
+            graph=graph,
+            handler=handler,
+            target=target,
+            dispatcher=dispatcher,
+            remove=True,
+        )
+    for added_item in event.added:
+        add_or_remove_notifiers(
+            object=added_item,
+            graph=graph,
+            handler=handler,
+            target=target,
+            dispatcher=dispatcher,
+            remove=False,
+        )

traits/observers/events.py

@@ -11,6 +11,9 @@
 """ Event objects received by change handlers added using observe.
 """
 
+from traits.observers._set_change_event import (   # noqa: F401
+    SetChangeEvent,
+)
 from traits.observers._trait_change_event import (   # noqa: F401
     TraitChangeEvent,
 )

traits/observers/tests/test_set_change_event.py

@@ -0,0 +1,66 @@
+# (C) Copyright 2005-2020 Enthought, Inc., Austin, TX
+# All rights reserved.
+#
+# This software is provided without warranty under the terms of the BSD
+# license included in LICENSE.txt and may be redistributed only under
+# the conditions described in the aforementioned license. The license
+# is also available online at http://www.enthought.com/licenses/BSD.txt
+#
+# Thanks for using Enthought open source!
+
+import unittest
+
+from traits.observers._set_change_event import (
+    SetChangeEvent,
+    set_event_factory,
+)
+from traits.trait_set_object import TraitSet
+
+
+class TestSetChangeEvent(unittest.TestCase):
+
+    def test_set_change_event_repr(self):
+        event = SetChangeEvent(
+            trait_set=set(),
+            added={1},
+            removed={3},
+        )
+        actual = repr(event)
+        self.assertEqual(
+            actual,
+            "SetChangeEvent(trait_set=set(), removed={3}, added={1})",
+        )
+
+
+class TestSetEventFactory(unittest.TestCase):
+    """ Test event factory compatibility with TraitSet.notify """
+
+    def test_trait_set_notification_compat(self):
+
+        events = []
+
+        def notifier(*args, **kwargs):
+            event = set_event_factory(*args, **kwargs)
+            events.append(event)
+
+        trait_set = TraitSet(
+            [1, 2, 3],
+            notifiers=[notifier],
+        )
+
+        # when
+        trait_set.add(4)
+
+        # then
+        event, = events
+        self.assertEqual(event.added, set([4]))
+        self.assertEqual(event.removed, set([]))
+
+        # when
+        events.clear()
+        trait_set.remove(4)
+
+        # then
+        event, = events
+        self.assertEqual(event.added, set([]))
+        self.assertEqual(event.removed, set([4]))