throttle - limiter

BaseLimiter

Bases: ABC

is_deny abstractmethod

is_deny(identifier, rules)

Checks if any rate limits are exceeded.

PARAMETER	DESCRIPTION
identifier	User or session identifier TYPE: str
rules	List of rate limit rules to check TYPE: list[LimitRule]

RETURNS	DESCRIPTION
list[float]	list[float]: List of waiting times until rate limits reset (empty if not exceeded)

Source code in webtool/throttle/limiter.py

@abstractmethod
def is_deny(self, identifier: str, rules: list[LimitRule]) -> list[float]:
    """
    Checks if any rate limits are exceeded.

    Parameters:
        identifier: User or session identifier
        rules: List of rate limit rules to check

    Returns:
        list[float]: List of waiting times until rate limits reset (empty if not exceeded)
    """

    raise NotImplementedError

RedisLimiter

RedisLimiter(redis_cache)

Bases: BaseLimiter

Rate limiter implementation using Redis for distributed rate limiting.

PARAMETER	DESCRIPTION
redis_cache	Redis client instance TYPE: RedisCache

Source code in webtool/throttle/limiter.py

def __init__(self, redis_cache: RedisCache):
    """
    Parameters:
         redis_cache: Redis client instance
    """

    self._cache = redis_cache.cache
    self._redis_function = self._cache.register_script(RedisLimiter._LUA_LIMITER_SCRIPT)
    self._json_encoder = ORJSONEncoder()
    self._json_decoder = ORJSONDecoder()

is_deny async

is_deny(identifier, rules)

Checks if any rate limits are exceeded.

PARAMETER	DESCRIPTION
identifier	User or session identifier TYPE: str
rules	List of rate limit rules to check TYPE: list[LimitRule]

RETURNS	DESCRIPTION
list[tuple[int, int, float]]	list[int, int, float]: List of (Limit Amount, Current, Time until rate limit reset (in seconds))

Source code in webtool/throttle/limiter.py

async def is_deny(self, identifier: str, rules: list[LimitRule]) -> list[tuple[int, int, float]]:
    """
    Checks if any rate limits are exceeded.

    Parameters:
        identifier: User or session identifier
        rules: List of rate limit rules to check

    Returns:
        list[int, int, float]: List of (Limit Amount, Current, Time until rate limit reset (in seconds))
    """

    ruleset = self._get_ruleset(identifier, rules)  # ruleset = {key: [limit, window_size], ...}
    result = await self._get_limits(ruleset)  # {key: [limit, amount, exist first request], ...}

    now = asyncio.get_running_loop().time()
    deny = [(val[0], val[1], float(val[2]) + ruleset[key][1] - now) for key, val in result.items()]

    return deny