A debounce function decorator in Python similar to the one in underscore.js, tested with 2.7

debounce.py

from threading import Timer


def debounce(wait):
    """ Decorator that will postpone a functions
        execution until after wait seconds
        have elapsed since the last time it was invoked. """
    def decorator(fn):
        def debounced(*args, **kwargs):
            def call_it():
                fn(*args, **kwargs)
            try:
                debounced.t.cancel()
            except(AttributeError):
                pass
            debounced.t = Timer(wait, call_it)
            debounced.t.start()
        return debounced
    return decorator

test_debounce.py

import unittest
import time
from debounce import debounce


class TestDebounce(unittest.TestCase):

    @debounce(10)
    def increment(self):
        """ Simple function that
            increments a counter when
            called, used to test the
            debounce function decorator """
        self.count += 1

    def setUp(self):
        self.count = 0

    def test_debounce(self):
        """ Test that the increment
            function is being debounced.
            The counter should only be incremented
            once 10 seconds after the last call
            to the function """
        self.assertTrue(self.count == 0)
        self.increment()
        self.increment()
        time.sleep(9)
        self.assertTrue(self.count == 0)
        self.increment()
        self.increment()
        self.increment()
        self.increment()
        self.assertTrue(self.count == 0)
        time.sleep(10)
        self.assertTrue(self.count == 1)

if __name__ == '__main__':
    unittest.main()

Here's another way. This uses Python 3's nonlocal keyword. The implementation is direct and very easy to understand.

import time


def debounce(s):
    """Decorator ensures function that can only be called once every `s` seconds.
    """
    def decorate(f):
        t = None

        def wrapped(*args, **kwargs):
            nonlocal t
            t_ = time.time()
            if t is None or t_ - t >= s:
                result = f(*args, **kwargs)
                t = time.time()
                return result
        return wrapped
    return decorate

Try it out.

@debounce(3)
def hi(name):
    print('hi {}'.format(name))


hi('dude')
time.sleep(1)
hi('mike')
time.sleep(1)
hi('mary')
time.sleep(1)
hi('jane')

@kylebebak your version is not "bounce" but as your description

ensures function that can only be called once every s seconds.

you version will always invoke the first call, and all the calls afterwards before the "s" seconds will be omitted

def debounce(wait):
    """ Decorator that will postpone a functions
        execution until after wait seconds
        have elapsed since the last time it was invoked. """

    def decorator(fn):
        def debounced(*args, **kwargs):
            def call_it():
                debounced._timer = None
                debounced._last_call = time.time()
                return fn(*args, **kwargs)

            time_since_last_call = time.time() - debounced._last_call
            if time_since_last_call >= wait:
                return call_it()

            if debounced._timer is None:
                debounced._timer = threading.Timer(wait - time_since_last_call, call_it)
                debounced._timer.start()

        debounced._timer = None
        debounced._last_call = 0

        return debounced

    return decorator

This debounce decorator will directly call the function if the last call is more than wait seconds ago. Otherwise it starts a timer to call the function after wait seconds. This means if the decorated function is called once every second, the first call is passed through and the second call (which happens 1s after the first call) is scheduled to be executed after 14s. Calls that happen after a call has been scheduled will be ignored. One could also change the decorator so that always the last given arguments are used for the next scheduled call.

@kylebebak your version is not "bounce" but as your description

ensures function that can only be called once every s seconds.

you version will always invoke the first call, and all the calls afterwards before the "s" seconds will be omitted

@neekey
Thanks for mentioning this, you're totally right! I think what I've written is typically called throttle, but I'm not sure this is the right name either.

@kylebebak here is a great visualization for understanding throttle vs debounce:
http://demo.nimius.net/debounce_throttle/

Didn't know about threading.Timer, I like this solution!

An optimization would be to protect the Timer creation using a threading.Lock. This would prevent problems with debounce being called between Timer's initialization and calling .start().

Another would be to use functools.wraps to preserve docstrings.

Hi everyone ! A few months ago I had the same need to have a working debounce annotation, after stumbling upon this discussion I created this open source project: https://github.com/salesforce/decorator-operations

The idea of the project is to regroup useful annotations such as debounce, throttle, filter... There are only 4 annotations available for now, but you're more than welcome to suggest new features or suggestions on how to improve the existing ones :D

👍 thanks for sharing

@KarlPatach that sounds good 👍

// feature 1
In addition to debounce and throttle
also has a situation, which can has both feature.

if calls too frequently, the executor may delayed indefinitely.
some states cannot be updated in time.
then wish debounce has a max delay time allow execute one time.
// I don't known how to call it, has a debounce time, and a max delay time

// feature 2
if asyncio version has supported, that will be nice.

// finally
I wonder is them thread safe

@KarlPatach That's awesome 🎉

My implementation (fully-typed and thread-safe):

import threading
from typing import Any, Callable, Optional, TypeVar, cast

class Debouncer:
    def __init__(self, f: Callable[..., Any], interval: float):
        self.f = f
        self.interval = interval
        self._timer: Optional[threading.Timer] = None
        self._lock = threading.Lock()

    def __call__(self, *args, **kwargs) -> None:
        with self._lock:
            if self._timer is not None:
                self._timer.cancel()
            self._timer = threading.Timer(self.interval, self.f, args, kwargs)
            self._timer.start()


VoidFunction = TypeVar("VoidFunction", bound=Callable[..., None])


def debounce(interval: float):
    """
    Wait `interval` seconds before calling `f`, and cancel if called again.
    The decorated function will return None immediately,
    ignoring the delayed return value of `f`.
    """

    def decorator(f: VoidFunction) -> VoidFunction:
        if interval <= 0:
            return f
        return cast(VoidFunction, Debouncer(f, interval))

    return decorator