Automatic unregistering of BindableProperty objects

nicegui/binding.py

@@ -1,5 +1,6 @@
 import asyncio
 import time
+import weakref
 from collections import defaultdict
 from collections.abc import Mapping
 from typing import Any, Callable, DefaultDict, Dict, Iterable, List, Optional, Set, Tuple, Union
@@ -10,7 +11,7 @@
 MAX_PROPAGATION_TIME = 0.01
 
 bindings: DefaultDict[Tuple[int, str], List] = defaultdict(list)
-bindable_properties: Dict[Tuple[int, str], Any] = {}
+bindable_properties: Dict[Tuple[int, str], weakref.finalize] = {}
 active_links: List[Tuple[Any, str, Any, str, Callable[[Any], Any]]] = []
 
 
@@ -149,11 +150,19 @@ def __set__(self, owner: Any, value: Any) -> None:
         if has_attr and not value_changed:
             return
         setattr(owner, '___' + self.name, value)
-        bindable_properties[(id(owner), self.name)] = owner
+        self._register(owner)
         _propagate(owner, self.name)
         if value_changed and self._change_handler is not None:
             self._change_handler(owner, value)
 
+    def _register(self, owner: Any) -> None:
+        registry_key = (id(owner), str(self.name))
+
+        def try_unregister() -> None:
+            bindable_properties.pop(registry_key, None)
+
+        bindable_properties.setdefault(registry_key, weakref.finalize(owner, try_unregister))
+
 
 def remove(objects: Iterable[Any]) -> None:
     """Remove all bindings that involve the given objects.
@@ -174,9 +183,11 @@ def remove(objects: Iterable[Any]) -> None:
         ]
         if not binding_list:
             del bindings[key]
-    for (obj_id, name), obj in list(bindable_properties.items()):
-        if id(obj) in object_ids:
-            del bindable_properties[(obj_id, name)]
+    for registry_key, finalizer in list(bindable_properties.items()):
+        obj_id, _ = registry_key
+        if obj_id in object_ids:
+            del bindable_properties[registry_key]
+            finalizer.detach()
 
 
 def reset() -> None:

tests/test_binding.py

@@ -1,8 +1,11 @@
+import gc
+import weakref
 from typing import Dict, Optional, Tuple
 
 from selenium.webdriver.common.keys import Keys
 
 from nicegui import ui
+from nicegui.binding import bindable_properties, BindableProperty, remove
 from nicegui.testing import Screen
 
 
@@ -105,3 +108,71 @@ def test_missing_target_attribute(screen: Screen):
 
     screen.open('/')
     screen.should_contain("text='Hello'")
+
+
+class TestBindablePropertyAutomaticCleanup:
+    @staticmethod
+    def make_label_bound_to_model(value: str) -> Tuple[ui.label, int, weakref.ref]:
+        class Model:
+            value = BindableProperty()
+
+            def __init__(self, value: str) -> None:
+                self.value = value
+
+        model = Model(value)
+        label = ui.label(model.value).bind_text(model, 'value')
+
+        return label, id(model), weakref.ref(model)
+
+    @staticmethod
+    def remove_bindings(*elements: ui.element) -> None:
+        remove(elements)  # usually the client calls this function on its elements when the user disconnects
+        gc.collect()  # should not really be necessary, but better safe than sorry
+
+    def test_model_automatic_cleanup(self, screen: Screen):
+        label, model_id, model_ref = self.make_label_bound_to_model('some value')
+
+        screen.open('/')
+        screen.should_contain('some value')
+
+        def model_is_alive() -> bool:
+            return model_ref() is not None
+
+        def model_has_bindings() -> bool:
+            return any(obj_id == model_id for obj_id, _ in bindable_properties)
+
+        assert model_is_alive()
+        assert model_has_bindings()
+
+        self.remove_bindings(label)
+
+        assert not model_is_alive()
+        assert not model_has_bindings()
+
+    def test_only_dead_model_unregistered(self, screen: Screen):
+        label_1, first_id, first_ref = self.make_label_bound_to_model('first')
+        _, second_id, second_ref = self.make_label_bound_to_model('second')
+
+        screen.open('/')
+        screen.should_contain('first')
+        screen.should_contain('second')
+
+        def is_alive(ref: weakref.ref) -> bool:
+            return ref() is not None
+
+        def has_bindings(owner: int) -> bool:
+            return any(obj_id == owner for obj_id, _ in bindable_properties)
+
+        assert is_alive(first_ref)
+        assert has_bindings(first_id)
+
+        assert is_alive(second_ref)
+        assert has_bindings(second_id)
+
+        self.remove_bindings(label_1)
+
+        assert not is_alive(first_ref)
+        assert not has_bindings(first_id)
+
+        assert is_alive(second_ref)
+        assert has_bindings(second_id)