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

Popular posts from this blog

Fixing the DeepSpeed Import Error While Fine-Tuning the Qwen Model

Amazon Linux 2023 - User data configuration for launch templates to connect to the EKS cluster

How to create ISM policy and rotate logs in opensearch