Add pause, resume and stop methods to Watcher (#174)

While a watcher is paused, changes to its dependencies do not
trigger a re-evaluation or callback. If a dependency changed while
the watcher was paused, then the watcher triggers once upon resume,
matching the behavior of Vue's watch API. The stop method is a named
equivalent of calling the watcher object directly.

Fixes #150

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: berendkleinhaneveld

README.md

@@ -27,7 +27,7 @@ Observe nested structures of dicts, lists, tuples and sets. Returns an observabl
 
 * `watcher = watch(func, callback, deep=False, immediate=False)`
 
-React to changes in the state accessed in `func` with `callback(old_value, new_value)`. Returns a watcher object. `del`elete it to disable the callback.
+React to changes in the state accessed in `func` with `callback(old_value, new_value)`. Returns a watcher object. `del`elete it to disable the callback. Use `watcher.pause()` and `watcher.resume()` to temporarily suspend the watcher: if a dependency changed while the watcher was paused, it triggers once upon resume. Call `watcher.stop()` (or the watcher object itself) to stop it permanently.
 
 * `wrapped_func = computed(func)`
 

observ/watcher.py

@@ -174,6 +174,8 @@ class Watcher(Generic[T]):
         "_deps",
         "_new_deps",
         "_number_of_callback_args",
+        "_paused",
+        "_pending_update",
         "_tasks",
         "callback",
         "callback_async",
@@ -206,6 +208,8 @@ def __init__(
         """
         self.id = next(_ids)
         self._active = True
+        self._paused = False
+        self._pending_update = False
         if callable(fn):
             if is_bound_method(fn):
                 self.fn = weak(fn.__self__, fn.__func__)
@@ -254,7 +258,16 @@ def __call__(self):
         any used resource. This can be used when special
         life-cycle management is needed for watchers.
         """
+        self.stop()
+
+    def stop(self) -> None:
+        """
+        Stop the watcher and clean up any used resource.
+        Equivalent to calling the watcher object directly.
+        """
         self._active = False
+        self._paused = False
+        self._pending_update = False
 
         # Clear resources
         self.fn = lambda: ()
@@ -265,6 +278,27 @@ def __call__(self):
         self._deps.clear()
         self._new_deps.clear()
 
+    def pause(self) -> None:
+        """
+        Temporarily pause the watcher: while paused, changes to
+        dependencies will not trigger a re-evaluation or callback.
+        Resume the watcher with `resume()`.
+        """
+        self._paused = True
+
+    def resume(self) -> None:
+        """
+        Resume the watcher after it was paused. If any of its
+        dependencies changed while the watcher was paused, it
+        will trigger once upon resume.
+        """
+        if not self._paused:
+            return
+        self._paused = False
+        if self._pending_update:
+            self._pending_update = False
+            self.update()
+
     def __del__(self):
         if Watcher.on_destroyed:
             Watcher.on_destroyed(self)
@@ -277,7 +311,19 @@ def active(self):
         """
         return self._active
 
+    @property
+    def paused(self):
+        """
+        Returns whether this watcher is currently paused.
+        Use `pause()` and `resume()` to control this state.
+        """
+        return self._paused
+
     def update(self) -> None:
+        if self._paused:
+            self._pending_update = True
+            return
+
         if self.lazy:
             self.dirty = True
             return
@@ -298,6 +344,11 @@ def run(self) -> None:
         # Early return for when the watcher has been deactivated
         if not self._active:
             return
+        # A watcher that was queued before it was paused should
+        # not run until it is resumed
+        if self._paused:
+            self._pending_update = True
+            return
         value = self.get()
         if self.deep or isinstance(value, Container) or value != self.value:
             old_value = self.value

tests/test_watcher.py

@@ -1,6 +1,6 @@
 from unittest.mock import Mock
 
-from observ import reactive, scheduler, watch
+from observ import computed, reactive, scheduler, watch
 
 
 def test_watcher_active(noop_request_flush):
@@ -85,3 +85,144 @@ def create_watchers(new):
     #     watcher.callback = None
 
     assert callback_args == [0]
+
+
+def test_watcher_stop(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback)
+
+    assert watcher.active
+
+    watcher.stop()
+
+    assert not watcher.active
+
+    a["foo"] = "baz"
+    scheduler.flush()
+
+    callback.assert_not_called()
+
+
+def test_watcher_pause_resume(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback)
+
+    assert not watcher.paused
+
+    watcher.pause()
+
+    assert watcher.paused
+
+    a["foo"] = "baz"
+    a["foo"] = "qux"
+    scheduler.flush()
+
+    # No callback while paused
+    callback.assert_not_called()
+
+    # Changes during the pause trigger the watcher once upon resume
+    watcher.resume()
+
+    assert not watcher.paused
+
+    scheduler.flush()
+    callback.assert_called_once_with("qux")
+
+
+def test_watcher_resume_without_changes(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback)
+
+    watcher.pause()
+    watcher.resume()
+    scheduler.flush()
+
+    callback.assert_not_called()
+
+    # Resuming a watcher that is not paused is a no-op
+    watcher.resume()
+    scheduler.flush()
+
+    callback.assert_not_called()
+
+
+def test_watcher_pause_resume_sync(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback, sync=True)
+
+    watcher.pause()
+
+    a["foo"] = "baz"
+
+    callback.assert_not_called()
+
+    # Sync watchers trigger immediately on resume
+    watcher.resume()
+
+    callback.assert_called_once_with("baz")
+
+
+def test_watcher_pause_while_queued(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback)
+
+    # Queue the watcher before pausing it
+    a["foo"] = "baz"
+    watcher.pause()
+    scheduler.flush()
+
+    callback.assert_not_called()
+
+    watcher.resume()
+    scheduler.flush()
+
+    callback.assert_called_once_with("baz")
+
+
+def test_watcher_stop_while_paused(noop_request_flush):
+    a = reactive({"foo": "bar"})
+    callback = Mock()
+
+    watcher = watch(lambda: a["foo"], callback)
+
+    watcher.pause()
+    a["foo"] = "baz"
+    watcher.stop()
+
+    assert not watcher.paused
+
+    watcher.resume()
+    scheduler.flush()
+
+    callback.assert_not_called()
+
+
+def test_computed_pause_resume(noop_request_flush):
+    a = reactive({"count": 1})
+
+    @computed
+    def double():
+        return a["count"] * 2
+
+    assert double() == 2
+
+    double.__watcher__.pause()
+
+    a["count"] = 2
+
+    # While paused, the computed expression is not invalidated
+    assert double() == 2
+
+    double.__watcher__.resume()
+
+    assert double() == 4