限速python装饰器

Posted 2023-02-23

【中文标题】限速python装饰器

【英文标题】：Rate-limiting python decorator

【发布时间】：2015-09-04 07:02:25

【问题描述】：

我发现this 基于redis 类的限速python 装饰器。如何编写一个类似的装饰器，只使用标准库中可用的东西，如下所示？

def ratelimit(limit, every):
    # ???? python magic ????

@ratelimit(limit=1, every=2)
def printlimited(x):
    print x

# print one number every two seconds
for x in range(10):
    printlimited(x)

*** 上还有其他answers，但它们不允许指定分母。

【参考方案1】：

您可以使用threading.Semaphore 来统计和阻止超出限制的请求，结合threading.Timer 来调度释放信号量的函数。

from threading import Semaphore, Timer
from functools import wraps

def ratelimit(limit, every):
    def limitdecorator(fn):
        semaphore = Semaphore(limit)
        @wraps(fn)
        def wrapper(*args, **kwargs):
            semaphore.acquire()
            try:
                return fn(*args, **kwargs)
            finally:                    # don't catch but ensure semaphore release
                timer = Timer(every, semaphore.release)
                timer.setDaemon(True)   # allows the timer to be canceled on exit
                timer.start()
        return wrapper
    return limitdecorator

我扩展了这个想法，并在 PyPI 上发布了一个名为 limit 的库。

【讨论】：

如果包装函数抛出异常，这不会释放信号量。您可以使用try..finally 子句来确保发生这种情况。