Python decorator for debouncing including function arguments

Question

How could one write a debounce decorator in python which debounces not only on function called but also on the function arguments/combination of function arguments used?

Debouncing means to supress the call to a function within a given timeframe, say you call a function 100 times within 1 second but you only want to allow the function to run once every 10 seconds a debounce decorated function would run the function once 10 seconds after the last function call if no new function calls were made. Here I'm asking how one could debounce a function call with specific function arguments.

An example could be to debounce an expensive update of a person object like:

@debounce(seconds=10)
def update_person(person_id):
    # time consuming, expensive op
    print('>>Updated person {}'.format(person_id))

Then debouncing on the function - including function arguments:

update_person(person_id=144)
update_person(person_id=144)
update_person(person_id=144)
>>Updated person 144

update_person(person_id=144)
update_person(person_id=355)
>>Updated person 144
>>Updated person 355

So calling the function update_person with the same person_id would be supressed (debounced) until the 10 seconds debounce interval has passed without a new call to the function with that same person_id.

There's a few debounce decorators but none includes the function arguments, example: https://gist.github.com/walkermatt/2871026

I've done a similar throttle decorator by function and arguments:

def throttle(s, keep=60):

    def decorate(f):

        caller = {}

        def wrapped(*args, **kwargs):
            nonlocal caller

            called_args = '{}'.format(*args)
            t_ = time.time()

            if caller.get(called_args, None) is None or t_ - caller.get(called_args, 0) >= s:
                result = f(*args, **kwargs)

                caller = {key: val for key, val in caller.items() if t_ - val > keep}
                caller[called_args] = t_
                return result

            # Keep only calls > keep
            caller = {key: val for key, val in caller.items() if t_ - val > keep}
            caller[called_args] = t_

        return wrapped

    return decorate

The main takaway is that it keeps the function arguments in caller[called_args]

See also the difference between throttle and debounce: http://demo.nimius.net/debounce_throttle/

Update:

After some tinkering with the above throttle decorator and the threading.Timer example in the gist, I actually think this should work:

from threading import Timer
from inspect import signature
import time


def debounce(wait):
    def decorator(fn):
        sig = signature(fn)
        caller = {}

        def debounced(*args, **kwargs):
            nonlocal caller

            try:
                bound_args = sig.bind(*args, **kwargs)
                bound_args.apply_defaults()
                called_args = fn.__name__ + str(dict(bound_args.arguments))
            except:
                called_args = ''

            t_ = time.time()

            def call_it(key):
                try:
                    # always remove on call
                    caller.pop(key)
                except:
                    pass

                fn(*args, **kwargs)

            try:
                # Always try to cancel timer
                caller[called_args].cancel()
            except:
                pass

            caller[called_args] = Timer(wait, call_it, [called_args])
            caller[called_args].start()

        return debounced

    return decorator

Answer 1

I've had the same need to build a debounce annotation for a personal project, after stumbling upon the same gist / discussion you have, I ended up with the following solution:

import threading
def debounce(wait_time):
    """
    Decorator that will debounce a function so that it is called after wait_time seconds
    If it is called multiple times, will wait for the last call to be debounced and run only this one.
    """

    def decorator(function):
        def debounced(*args, **kwargs):
            def call_function():
                debounced._timer = None
                return function(*args, **kwargs)
            # if we already have a call to the function currently waiting to be executed, reset the timer
            if debounced._timer is not None:
                debounced._timer.cancel()

            # after wait_time, call the function provided to the decorator with its arguments
            debounced._timer = threading.Timer(wait_time, call_function)
            debounced._timer.start()

        debounced._timer = None
        return debounced

    return decorator

I've created an open-source project to provide functions such as debounce, throttle, filter ... as decorators, contributions are more than welcome to improve on the solution I have for these decorators / add other useful decorators: decorator-operations repository