用于去抖动（包括函数参数）的 Python 装饰器

如何在 python 中编写一个去抖装饰器，它不仅对调用的函数进行去抖，而且对所使用的函数参数/函数参数组合进行去抖？

去抖意味着在给定的时间范围内抑制对函数的调用，假设您在 1 秒内调用一个函数 100 次，但您只想允许该函数每 10 秒运行一次，去抖装饰函数将在 10 秒后运行该函数一次如果没有进行新的函数调用，则为最后一次函数调用。在这里我问的是如何使用特定的函数参数来消除函数调用的反跳。

一个例子是取消对人员对象的昂贵更新，例如：

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

然后对函数进行去抖动 - 包括函数参数：

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

因此，使用相同的 person_id 调用函数 update_person 将被抑制（去抖），直到 10 秒的去抖间隔过去，而没有对具有相同 person_id 的函数进行新的调用。

有一些去抖装饰器，但没有一个包含函数参数，例如：https://gist.github.com/walkermatt/2871026

我通过函数和参数做了一个类似的节流装饰器：

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

主要要点是它将函数参数保留在 call[used args] 中

另请参阅节流阀和防抖之间的区别：http://demo.nimius.net/debounce_throttle/

Update:

在对上面的节流阀装饰器和要点中的 threading.Timer 示例进行一些修改之后，我实际上认为这应该有效：

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

在偶然发现之后，我也有同样的需要为个人项目构建去抖注释相同的要点/讨论你有，我最终得到了以下解决方案：

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

我创建了一个开源项目来提供诸如去抖、节流、过滤等功能...作为装饰器，非常欢迎贡献来改​​进我为这些装饰器提供的解决方案/添加其他有用的装饰器：装饰器操作存储库