Source code for pyqqq.utils.limiter
import time
_MAX_CALLS = 5
_PERIOD = 1
_SCOPE = 'default'
[docs]
class CallLimiter:
"""
API 호출 제한을 관리하기 위한 싱글턴 클래스입니다.
이 클래스는 API 호출 빈도를 제한하여 너무 많은 요청으로 인해 서비스 제한을 받는 것을 방지합니다.
이 클래스의 인스턴스는 전역에서 단 하나만 존재하며, 다양한 'scope'에 대한 호출 제한 윈도우를 관리합니다.
"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(CallLimiter, cls).__new__(cls)
cls._instance.windows = {'default': []}
return cls._instance
[docs]
def wait_limit_rate(self, max_calls: int = _MAX_CALLS, period: float = _PERIOD, scope: str = _SCOPE):
"""
지정된 시간 간격 동안 최대 호출 횟수를 초과하지 않도록 기다립니다.
호출 윈도우를 관리하여 max_calls 및 period를 기반으로 호출이 허용될 때까지 대기합니다.
'scope'에 따라 다른 API 또는 함수 호출에 대해 다른 제한을 설정할 수 있습니다.
Args:
max_calls (int): 지정된 시간 동안 허용되는 최대 호출 횟수.
period (float): 호출 제한을 적용할 시간 간격(초).
scope (str): 호출 제한을 적용할 범위 또는 이름.
Examples:
>>> call_limiter = CallLimiter()
>>> call_limiter.wait_limit_rate(10, 60, 'api_calls')
"""
if scope not in self.windows:
self.windows[scope] = []
now = time.time()
window = list(filter(lambda x: now - x < period, self.windows[scope]))
window_size = len(window)
if window_size >= max_calls:
time.sleep(period)
elif window_size > 0:
time.sleep(period / max_calls)
window.append(time.time())
self.windows[scope] = window
[docs]
def limit_calls(max_calls: int = _MAX_CALLS, period: float = _PERIOD, scope: str = _SCOPE):
"""
여러 함수에 대한 호출을 제한하는 데코레이터입니다.
Args:
max_calls (int): 최대 호출 횟수
period (float): 기간(초) 동안의 호출 횟수를 제한합니다.
scope (str): 호출 제한을 적용할 스코프 이름입니다.
"""
def decorator(func):
def wrapper(*args, **kwargs):
CallLimiter().wait_limit_rate(max_calls=max_calls, period=period, scope=scope)
result = func(*args, **kwargs)
return result
return wrapper
return decorator