Store weak references to bound methods (#94)

* Store weak references to bound methods

* Cleanup

* Improve doc string

* Change NotImplementedError to WrongNumberOfArgumentsError and update PySide test to xfail

observ/watcher.py

@@ -10,7 +10,7 @@
 import inspect
 from itertools import count
 from typing import Any, Callable, Optional, TypeVar
-from weakref import WeakSet
+from weakref import ref, WeakSet
 
 from .dep import Dep
 from .observables import DictProxyBase, ListProxyBase, Proxy, SetProxyBase
@@ -124,7 +124,10 @@ def __init__(
         """
         self.id = next(_ids)
         if callable(fn):
-            self.fn = fn
+            if is_bound_method(fn):
+                self.fn = weak(fn.__self__, fn.__func__)
+            else:
+                self.fn = fn
         else:
             self.fn = lambda: fn
             # Default to deep watching when watching a proxy
@@ -134,7 +137,10 @@ def __init__(
         self._deps, self._new_deps = WeakSet(), WeakSet()
 
         self.sync = sync
-        self.callback = callback
+        if is_bound_method(callback):
+            self.callback = weak(callback.__self__, callback.__func__)
+        else:
+            self.callback = callback
         self.deep = bool(deep)
         self.lazy = lazy
         self.dirty = self.lazy
@@ -252,3 +258,52 @@ def depend(self) -> None:
     @property
     def fn_fqn(self) -> str:
         return f"{self.fn.__module__}.{self.fn.__qualname__}"
+
+
+def weak(obj: Any, method: Callable):
+    """
+    Returns a wrapper for the given method that will only call the method if the
+    given object is not garbage collected yet. It does so by using a weakref.ref
+    and checking its value before calling the actual method when the wrapper is
+    called.
+    """
+    weak_obj = ref(obj)
+
+    sig = inspect.signature(method)
+    nr_arguments = len(sig.parameters)
+
+    if nr_arguments == 1:
+
+        @wraps(method)
+        def wrapped():
+            if this := weak_obj():
+                return method(this)
+
+        return wrapped
+    elif nr_arguments == 2:
+
+        @wraps(method)
+        def wrapped(new):
+            if this := weak_obj():
+                return method(this, new)
+
+        return wrapped
+    elif nr_arguments == 3:
+
+        @wraps(method)
+        def wrapped(new, old):
+            if this := weak_obj():
+                return method(this, new, old)
+
+        return wrapped
+    else:
+        raise WrongNumberOfArgumentsError(
+            "Please use 1, 2 or 3 arguments for callbacks"
+        )
+
+
+def is_bound_method(fn: Callable):
+    """
+    Returns whether the given function is a bound method.
+    """
+    return hasattr(fn, "__self__") and hasattr(fn, "__func__")

tests/test_references.py

@@ -0,0 +1,155 @@
+from weakref import ref
+
+from PySide6 import QtWidgets
+import pytest
+
+from observ import reactive, watch
+from observ.watcher import WrongNumberOfArgumentsError
+
+
+def test_no_strong_reference_to_callback():
+    class Counter:
+        count_a = 0
+        count_b = 0
+        count_c = 0
+
+        def count(self):
+            type(self).count_a += 1
+
+        def count_new(self, new):
+            type(self).count_b = new
+
+        def count_new_old(self, new, old):
+            type(self).count_c += new - (old or 0)
+
+    state = reactive({"count": 0})
+
+    counter = Counter()
+
+    counter.watcher_a = watch(
+        lambda: state["count"],
+        counter.count,
+        deep=True,
+        sync=True,
+    )
+    counter.watcher_b = watch(
+        lambda: state["count"],
+        counter.count_new,
+        deep=True,
+        sync=True,
+    )
+    counter.watcher_c = watch(
+        lambda: state["count"],
+        counter.count_new_old,
+        deep=True,
+        sync=True,
+    )
+
+    state["count"] += 1
+
+    assert Counter.count_a == 1
+    assert Counter.count_b == 1
+    assert Counter.count_c == 1
+
+    del counter
+
+    state["count"] += 1
+
+    assert Counter.count_a == 1
+    assert Counter.count_b == 1
+    assert Counter.count_c == 1
+
+
+def test_no_strong_reference_to_fn():
+    class Counter:
+        def __init__(self, state):
+            self.state = state
+
+        def count(self):
+            return self.state["count"]
+
+    count = 0
+
+    def cb():
+        nonlocal count
+        count += 1
+
+    state = reactive({"count": 0})
+    counter = Counter(state)
+
+    counter.watcher = watch(
+        counter.count,
+        cb,
+        deep=True,
+        sync=True,
+    )
+
+    state["count"] += 1
+
+    assert count == 1
+
+    del counter
+
+    state["count"] += 1
+
+    assert count == 1
+
+
+def test_check_nr_arguments_of_weak_callback():
+    class Counter:
+        def cb(self, new, old, too_much):
+            pass
+
+    counter = Counter()
+    state = reactive({})
+
+    with pytest.raises(WrongNumberOfArgumentsError):
+        watch(
+            state,
+            counter.cb,
+            sync=True,
+            deep=True,
+        )
+
+
+@pytest.mark.xfail
+def test_qt_integration(qapp):
+    class Label(QtWidgets.QLabel):
+        count = 0
+
+        def __init__(self, state):
+            super().__init__()
+            self.state = state
+
+            self.watcher = watch(
+                self.count_display,
+                # Use a method from QLabel directly as callback. These methods don't
+                # have a __func__ attribute, so currently this won't be wrapped by
+                # the 'weak' wrapper in the watcher, so it will create a strong
+                # reference instead to self.
+                # Ideally we would be able to detect this and wrap it, but then
+                # there is the problem that these kinds of methods don't have a
+                # signature, so we can't detect the number of arguments to supply.
+                # I guess we can assume that we should supply only two arguments
+                # in those cases: 'self' and 'new'?
+                # We'll solve this if this becomes an actual problem :)
+                self.setText,
+                deep=True,
+                sync=True,
+            )
+
+        def count_display(self):
+            return str(self.state["count"])
+
+    state = reactive({"count": 0})
+    label = Label(state)
+
+    state["count"] += 1
+
+    assert label.text() == "1"
+
+    weak_label = ref(label)
+
+    del label
+
+    assert not weak_label()