Decorator in python
def distributed_rate_limit(calls: int, period_in_sec: int, counter: Counter):
def outer_wrapper(func):
@wraps(func)
def inner_wrapper(*args, **kwargs):
counter.lock.acquire()
now = int(datetime.now().timestamp())
start_time = get_cache_int_value(counter.timer_key, now)
range_time = now - start_time
# checking range time and reset request count
if range_time >= period_in_sec or range_time == 0:
redis.set(counter.counter_key, 0)
redis.set(counter.timer_key, now)
count = get_cache_int_value(counter.counter_key, 0)
# checking limit calls
if count < calls:
redis.set(counter.counter_key, count + 1)
counter.lock.release()
return func(*args, **kwargs)
counter.lock.release()
raise TooManyRequestException("Too many requests exception")
return inner_wrapper
return outer_wrapper
Comments
Post a Comment