如何根据方法参数应用速率限制

Question

我正在使用 python 模块

ratelimit

来限制调用 rest api 的函数，我需要根据请求的方法应用限制，例如每 10 秒

PUT/POST/DELETE

1 次，每 1 秒

GET

5 次，我怎样才能在不将功能分成两部分的情况下实现这一目标？

from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=1 if method != 'GET' else 5), period=10 if method != 'GET' else 1)
def callrest(method, url, data):
    ...

可以这样做吗？

Answer 1

这是一个不需要装饰器就可以使用的“速率限制器”，因此您可以创建两个单独的速率限制器，并根据调用的方法使用一个或另一个。此类的实例可以跨多个线程使用，因为它在内部使用锁来保持其内部状态一致。您还可以使用此类的 managed 版本跨多个进程使用。

第一节课：

from multiprocessing.managers import BaseManager
from collections import deque
from threading import Lock
import time

class RateLimiter:
    def __init__(self, call_count, period=1.0):
        self._call_count = int(call_count)
        self._period = float(period)
        self._called_timestamps = deque()
        self._lock = Lock()

    def throttle(self):
        with self._lock:
            while True:
                now = time.monotonic()
                while self._called_timestamps:
                    time_left = self._called_timestamps[0] + self._period - now
                    if time_left >= 0:
                        break
                    self._called_timestamps.popleft()
                if len(self._called_timestamps) < self._call_count:
                    break
                time.sleep(time_left)
            self._called_timestamps.append(now)

# A "managed" RateLimiter is required for use with multiprocessing:
class RateLimiterManager(BaseManager):
    pass

RateLimiterManager.register('RateLimiter', RateLimiter)

然后你的代码将修改如下：

get_rate_limiter = RateLimiter(5, 1.0)
put_post_delete_rate_limiter = RateLimiter(1, 10.0)

def callrest(method, url, data):
    rate_limiter = get_rate_limiter if method == 'GET' else put_post_delete_rate_limiter
    rate_limiter.throttle()
    ...

Answer 2

只是在想这是否可以帮助您：

from ratelimit import limits, sleep_and_retry

def get_rate_limit(method):
    if method in ['PUT', 'POST', 'DELETE']:
        return (1, 10)
    elif method == 'GET':
        return (5, 1)

def throttle(method):
    calls, period = get_rate_limit(method)
    return limits(calls=calls, period=period)

@sleep_and_retry
@throttle(method)
def callrest(method, url, data):
    # function implementation here

Answer 3

当

ratelimit

已经运行良好时，您不必通过创建自己的速率限制器来重新发明轮子。

而是先将

ratelimit.limits

返回的装饰器对象存储到变量中，用变量装饰

callrest

函数，然后在函数内修改

clamped_calls

属性（存储

calls的值

参数）和变量的

period

属性根据

method

参数的值：

from ratelimit import limits, sleep_and_retry

my_limits = limits()

@sleep_and_retry
@my_limits
def callrest(method, url, data):
    if method == 'GET':
        my_limits.clamped_calls = 5
        my_limits.period = 1
    else:
        my_limits.clamped_calls = 1
        my_limits.period = 10
    ...

演示：https://replit.com/@blhsing/TriangularPerfectSdk