Clock: Enhance and fix @triggered decorator (instance isolation, debouncing, lazy initialization) and add deterministic pytest fixture for manual clock advancement (kivy#9275)

* fix and improve triggered decorator

* fix pep8

* add kivy_clock_advance and replace time.sleep in tests

* fix pep8

Author: DexerBR

doc/sources/migration.rst

@@ -875,6 +875,68 @@ You no longer need to manually clean up groups:
 +---------------------------+---------------------------+----------------------------------------+
 
 
+=====
+Clock
+=====
+
+*Improved @triggered Decorator Behavior, Instance Isolation and Debouncing*
+
+In Kivy 3.x.x, the :func:`~kivy.clock.triggered` decorator has been significantly
+improved. Previously, when used as a method decorator, it shared a single trigger
+and state across all instances of a class. This meant that calling the method on
+one instance would throttle calls on all other instances, and arguments from
+different instances could overwrite each other.
+
+**Behavior Changes**
+
+* Improved **Instance Isolation**: Each instance now has its own isolated trigger and
+  argument storage. Calling a triggered method on ``widget_a`` no longer affects
+  ``widget_b``.
+* Improved **Lazy Initialization**: Triggers are now created only when the decorated
+  function is first called, improving initialization performance.
+* New **is_triggered Property**: A new ``is_triggered`` property was added to the decorated
+  function/method, allowing you to check if a call is currently pending.
+* New **debounce Parameter**: A new ``debounce=False`` (default) parameter was
+  added.
+
+  * **Throttling** (default): Subsequent calls while a trigger is active
+    update the arguments but do *not* reset the timer. The function fires once
+    after the initial timeout.
+  * **Debouncing** (``debounce=True``): Subsequent calls cancel any pending
+    execution and reschedule it. The function only fires after the caller
+    stops calling it for the duration of the ``timeout``.
+
+**Migration Impact**
+
+This is primarily a **bug fix** and a set of **new features**. It should not
+require code changes for most applications. However, if your codebase
+intentionally relied on the legacy shared throttling behavior across different
+instances, you can restore this behavior by using the **``@classmethod``** 
+decorator above ``@triggered``. 
+
+This ensures the trigger is bound to the class object rather than individual 
+instances, restoring the shared behavior in an idiomatic way.
+
+.. code-block:: python
+
+    class MyWidget(Widget):
+        # Default in 3.x.x: Isolated per instance
+        @triggered(0.1)
+        def sync_ui(self, *args):
+            pass
+
+        # Shared behavior (same as legacy 2.x.x): Shared by all instances
+        @classmethod
+        @triggered(0.1)
+        def sync_shared_data(cls, *args):
+            pass
+
+        # Optional: Debouncing (0.1s from the LAST call)
+        @triggered(0.1, debounce=True)
+        def search_input(self, text):
+            pass
+
+
 Application Storage Directories
 ================================
 

kivy/clock.py

@@ -456,7 +456,8 @@ def handle_exception(self, inst):
 
 from sys import platform
 from os import environ
-from functools import wraps, partial
+from functools import wraps, partial, update_wrapper
+from weakref import WeakKeyDictionary, ref as weakref_ref
 from kivy.context import register_context
 from kivy.config import Config
 from kivy.logger import Logger
@@ -1094,62 +1095,227 @@ def callback_func(dt):
     return delayed_func
 
 
-def triggered(timeout=0, interval=False):
-    '''Decorator that will trigger the call of the function at the specified
-    timeout, through the method :meth:`CyClockBase.create_trigger`. Subsequent
-    calls to the decorated function (while the timeout is active) are ignored.
+# The @triggered decorator uses two classes to handle the duality of Python
+# functions (global functions vs. instance methods):
+#
+# 1. _TriggeredWrapper (The Descriptor):
+#    - Lives at the module level (global) or class level.
+#    - For global functions: Manages its own `_trigger` and state in __call__.
+#    - For class methods: Implements the Descriptor Protocol (__get__) to
+#      intercept access via an instance (e.g., `widget.method()`).
+#    - Uses a WeakKeyDictionary to map instances to their respective
+#      _BoundTrigger objects, ensuring no memory leaks.
+#
+# 2. _BoundTrigger (The Instance State):
+#    - The object returned when accessing a method on an instance.
+#    - Maintains a `Clock.create_trigger` and argument buffers
+#      (`_args`, `_kwargs`) that are fully ISOLATED for each instance of the
+#      widget/object.
+#    - This isolation allows debouncing/throttling to work independently:
+#      triggering on 'Widget A' does not affect or cancel the schedule of
+#      'Widget B'.
+#
+# This separation resolves "cross-talk" between instances while keeping
+# the decorator flexible enough for use outside of classes.
+
+
+class _TriggeredWrapper:
+    def __init__(self, func, timeout, interval, debounce):
+        self._func = func
+        self._timeout = timeout
+        self._interval = interval
+        self._debounce = debounce
+        self._args = []
+        self._kwargs = {}
+        self._trigger = None
+        self._instances = WeakKeyDictionary()
+        update_wrapper(self, func, updated=())
 
-    It can be helpful when an expensive function (i.e. call to a server) can be
-    triggered by different methods. Setting a proper timeout will delay the
-    calling and only one of them will be triggered.
+    @property
+    def is_triggered(self):
+        """Returns True if the trigger is currently scheduled."""
+        if self._trigger is not None:
+            return self._trigger.is_triggered
+        return False
+
+    def _ensure_trigger(self):
+        if self._trigger is None:
+            self._trigger = Clock.create_trigger(
+                self.cb_function, timeout=self._timeout, interval=self._interval
+            )
+
+    def cb_function(self, dt):
+        self._func(*tuple(self._args), **self._kwargs)
+
+    def __call__(self, *args, **kwargs):
+        self._ensure_trigger()
+        if self._debounce:
+            self._trigger.cancel()
+        self._args[:] = args
+        self._kwargs.clear()
+        self._kwargs.update(kwargs)
+        self._trigger()
+
+    def cancel(self):
+        if self._trigger is not None:
+            self._trigger.cancel()
+
+    def __get__(self, instance, owner):
+        if instance is None:
+            return self
+
+        if instance not in self._instances:
+            # Create a bound version for this instance
+            inst_weak = weakref_ref(instance)
+            inst_args = []
+            inst_kwargs = {}
+            func = self._func
+            timeout = self._timeout
+            interval = self._interval
+            debounce = self._debounce
+
+            def inst_cb(dt):
+                inst = inst_weak()
+                if inst is not None:
+                    func(inst, *tuple(inst_args), **inst_kwargs)
+
+            inst_trigger = Clock.create_trigger(
+                inst_cb, timeout=timeout, interval=interval
+            )
+
+            bound_trigger = _BoundTrigger(
+                inst_trigger, inst_args, inst_kwargs, inst_weak, debounce
+            )
+            update_wrapper(bound_trigger, func)
+            self._instances[instance] = bound_trigger
+
+        return self._instances[instance]
+
+
+class _BoundTrigger(object):
+
+    def __init__(self, trigger, args, kwargs, inst_weak, debounce):
+        self._trigger = trigger
+        self._args = args
+        self._kwargs = kwargs
+        self._inst_weak = inst_weak
+        self._debounce = debounce
+
+    def __call__(self, *args, **kwargs):
+        if self._inst_weak() is not None:
+            if self._debounce:
+                self._trigger.cancel()
+            self._args[:] = args
+            self._kwargs.clear()
+            self._kwargs.update(kwargs)
+            self._trigger()
 
-        @triggered(timeout, interval=False)
-        def callback(id):
-            print('The callback has been called with id=%d' % id)
+    @property
+    def is_triggered(self):
+        """Returns True if the trigger is currently scheduled."""
+        return self._trigger.is_triggered
 
-        >> callback(id=1)
-        >> callback(id=2)
-        The callback has been called with id=2
+    def cancel(self):
+        """Unschedule the trigger."""
+        self._trigger.cancel()
 
-    The decorated callback can also be unscheduled using:
 
-        >> callback.cancel()
+def triggered(timeout=0, interval=False, debounce=False):
+    """Decorator that schedules the execution of a function after a specified
+    timeout using :meth:`CyClockBase.create_trigger`.
 
-    .. versionadded:: 1.10.1
-    '''
-    fun = None
+    This decorator provides several key behaviors:
 
-    if callable(timeout):
-        fun = timeout
-        timeout = 0
+    * **Throttling** (default): Subsequent calls while a trigger is active are
+      ignored. The function fires once after the initial timeout.
+    * **Debouncing**: If ``debounce=True``, subsequent calls cancel any
+      pending execution and reschedule it. The function only fires after the
+      caller stops calling it for the duration of the ``timeout``.
+    * **Last Arguments Win**: If called multiple times before the trigger
+      fires, the arguments from the **latest** call are used.
+    * **Instance Isolation**: When used as a method decorator, each instance
+      gets its own isolated trigger and state.
+    * **Thread Safety**: Safe to call from external threads.
 
-    def wrapper_triggered(func):
+    The decorated function gains a ``.cancel()`` method to unschedule any
+    pending execution, and an ``.is_triggered`` property to check its status.
+
+    :param timeout: The delay (in seconds) before the function is called.
+    :type timeout: float, defaults to 0
+    :param interval: If True, the trigger will be repeating (standard Clock
+        interval behavior).
+    :type interval: bool, defaults to False
+    :param debounce: If True, subsequent calls will cancel and reschedule the
+        trigger.
+    :type debounce: bool, defaults to False
+
+    Example of a global function without debouncing::
+
+        @triggered(0.1)
+        def sync_data(user_id):
+            print(f"Syncing data for {user_id}")
+
+        sync_data(1)
+        sync_data(2)  # Overwrites arguments of the first call
+        # 0.1s later: "Syncing data for 2"
 
-        _args = []
-        _kwargs = {}
+    Example of a global function with debouncing::
 
-        def cb_function(dt):
-            func(*tuple(_args), **_kwargs)
+        @triggered(0.1, debounce=True)
+        def sync_data(user_id):
+            print(f"Syncing data for {user_id}")
 
-        cb_trigger = Clock.create_trigger(
-            cb_function,
-            timeout=timeout,
-            interval=interval)
+        sync_data(1)
+        # 0.05s later
+        sync_data(2)  # Cancels the first call and reschedules for 0.1s from now
+        # 0.1s after the LAST call ("sync_data(2)"): "Syncing data for 2"
 
-        @wraps(func)
-        def trigger_function(*args, **kwargs):
-            _args[:] = []
-            _args.extend(list(args))
-            _kwargs.clear()
-            _kwargs.update(kwargs)
-            cb_trigger()
+    Example of an instance method::
 
-        def trigger_cancel():
-            cb_trigger.cancel()
+        class DataMapper:
+            @triggered(0.2)
+            def refresh(self, *args):
+                # This will be throttled per-instance
+                pass
 
-        setattr(trigger_function, 'cancel', trigger_cancel)
+        dm1, dm2 = DataMapper(), DataMapper()
+        dm1.refresh() # Scheduled
+        dm2.refresh() # Scheduled independently
 
-        return trigger_function
+    Example with ``classmethod`` and ``staticmethod`` (always put
+    ``@triggered`` below them)::
+
+        class Utils:
+            @classmethod
+            @triggered(0.1)
+            def global_refresh(cls, *args):
+                # shared by all instances
+                pass
+
+            @staticmethod
+            @triggered(0.1, debounce=True)
+            def log_event(message):
+                pass
+
+    To cancel a pending call::
+
+        sync_data.cancel()
+
+    .. versionadded:: 1.10.1
+    .. versionchanged:: 3.0.0
+        Fixed behavior for instance methods to ensure state isolation between
+        different objects. Added ``debounce`` parameter.
+    """
+    fun = None
+
+    # handle both shorthand usage (@triggered)
+    # and parameterized usage (@triggered(0.1))
+    if callable(timeout):
+        fun = timeout
+        timeout = 0
+
+    def wrapper_triggered(func):
+        return _TriggeredWrapper(func, timeout, interval, debounce)
 
     if fun is not None:
         return wrapper_triggered(fun)

kivy/tests/conftest.py

@@ -4,8 +4,8 @@
 kivy_eventloop = os.environ.get('KIVY_EVENTLOOP', 'asyncio')
 
 try:
-    from .fixtures import kivy_app, kivy_clock, kivy_metrics, \
-        kivy_exception_manager
+    from .fixtures import kivy_app, kivy_clock, kivy_clock_advance, \
+        kivy_metrics, kivy_exception_manager
 except (SyntaxError, ImportError):
     # async app tests would be skipped due to async_run forcing it to skip so
     # it's ok to fail here as it won't be used anyway

kivy/tests/fixtures.py

@@ -17,7 +17,8 @@
 import time
 import os.path
 
-__all__ = ('kivy_clock', 'kivy_metrics', 'kivy_exception_manager', 'kivy_app',
+__all__ = ('kivy_clock', 'kivy_clock_advance', 'kivy_metrics',
+           'kivy_exception_manager', 'kivy_app',
            'kivy_init')
 
 @pytest.fixture()
@@ -91,6 +92,28 @@ def kivy_clock():
         context.pop()
 
 
+@pytest.fixture()
+def kivy_clock_advance(kivy_clock):
+    """A fixture that provides a helper to advance the Clock
+       deterministically by mocking its time and ticking it.
+       Usage:
+           def test_foo(kivy_clock_advance):
+               kivy_clock_advance(0.1)  # advances the clock by 0.1s
+    """
+    class ClockController:
+        def __init__(self, clock):
+            self.clock = clock
+            self.now = 100.0
+            clock.time = lambda: self.now
+            clock._last_tick = self.now
+
+        def __call__(self, secs):
+            self.now += secs
+            self.clock.tick()
+
+    return ClockController(kivy_clock)
+
+
 @pytest.fixture()
 def kivy_metrics():
     from kivy.context import Context