throttle - decorator

THROTTLE_RULE_ATTR_NAME module-attribute

THROTTLE_RULE_ATTR_NAME = '_throttle_rules'

LimitRule

LimitRule(
    max_requests, interval, throttle_key, method, scopes
)

Represents a single rate limiting rule.

Contains the conditions and parameters for a rate limit: - Maximum requests allowed - Time interval - Unique identifier - HTTP methods (optional) - User scopes (optional)

PARAMETER	DESCRIPTION
max_requests	Maximum number of requests allowed TYPE: int
interval	Time interval in seconds TYPE: int
throttle_key	Unique identifier for this rule TYPE: str
method	List of HTTP methods this rule applies to TYPE: Optional[list[str]]
scopes	List of user scopes this rule applies to TYPE: Optional[list[str]]

Source code in webtool/throttle/decorator.py

def __init__(
    self,
    max_requests: int,
    interval: int,
    throttle_key: str,
    method: Optional[list[str]],
    scopes: Optional[list[str]],
):
    """
    Parameters:
        max_requests: Maximum number of requests allowed
        interval: Time interval in seconds
        throttle_key: Unique identifier for this rule
        method: List of HTTP methods this rule applies to
        scopes: List of user scopes this rule applies to
    """

    self.max_requests: int = max_requests
    self.interval: int = interval
    self.throttle_key: str = throttle_key
    self.method: Optional[set[str]] = set(method)
    self.scopes: Optional[set[str]] = set(scopes)
    self.for_user: bool = "user" in scopes or ("user" in scopes) == ("anno" in scopes)
    self.for_anno: bool = "anno" in scopes or ("user" in scopes) == ("anno" in scopes)

    self.scopes.discard("user")
    self.scopes.discard("anno")

max_requests instance-attribute

max_requests = max_requests

interval instance-attribute

interval = interval

throttle_key instance-attribute

throttle_key = throttle_key

method instance-attribute

method = set(method)

scopes instance-attribute

scopes = set(scopes)

for_user instance-attribute

for_user = (
    "user" in scopes or "user" in scopes == "anno" in scopes
)

for_anno instance-attribute

for_anno = (
    "anno" in scopes or "user" in scopes == "anno" in scopes
)

is_enabled

is_enabled(scope, is_user, auth_data)

Checks if this rule should be applied based on request context.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict
is_user	True if user, False unauthorized TYPE: bool
auth_data	webtool.auth.models.AuthData TYPE: AuthData

RETURNS	DESCRIPTION
bool	indicating if rule should be applied TYPE: bool

Source code in webtool/throttle/decorator.py

def is_enabled(
    self,
    scope: dict,
    is_user: bool,
    auth_data: AuthData,
) -> bool:
    """
    Checks if this rule should be applied based on request context.

    Parameters:
        scope: ASGI request scope
        is_user: True if user, False unauthorized
        auth_data: webtool.auth.models.AuthData

    Returns:
        bool: indicating if rule should be applied
    """

    scopes = auth_data.data.get("scope") if auth_data.data else []

    if self.method and (scope.get("method") not in self.method):
        return False

    if self.scopes and not self.scopes.intersection(set(scopes)):
        return False

    if is_user:
        return self.for_user

    return self.for_anno

LimitRuleManager

LimitRuleManager()

Container class for managing multiple rate limit rules. Allows adding and checking rules for specific requests.

Source code in webtool/throttle/decorator.py

def __init__(self):
    self.rules: set[LimitRule] = set()

rules instance-attribute

rules = set()

should_limit

should_limit(scope, is_user, auth_data)

Determines which rules should be applied for a given request.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict
is_user	True if user, False unauthorized TYPE: bool
auth_data	webtool.auth.models.AuthData TYPE: AuthData

RETURNS	DESCRIPTION
list[LimitRule]	List of applicable rules

Source code in webtool/throttle/decorator.py

def should_limit(
    self,
    scope: dict,
    is_user: bool,
    auth_data: AuthData,
) -> list[LimitRule]:
    """
    Determines which rules should be applied for a given request.

    Parameters:
        scope: ASGI request scope
        is_user: True if user, False unauthorized
        auth_data: webtool.auth.models.AuthData

    Returns:
        List of applicable rules
    """

    rules = [rule for rule in self.rules if rule.is_enabled(scope, is_user, auth_data)]

    return rules

add_rules

add_rules(rule)

Adds a new rate limit rule to the collection.

PARAMETER	DESCRIPTION
rule	LimitRule instance to add TYPE: LimitRule

Source code in webtool/throttle/decorator.py

def add_rules(self, rule: LimitRule) -> None:
    """
    Adds a new rate limit rule to the collection.

    Parameters:
        rule: LimitRule instance to add
    """

    self.rules.add(rule)

limiter

limiter(
    max_requests,
    interval=3600,
    throttle_key=None,
    method=None,
    scopes=None,
)

Decorator for implementing rate limiting on functions.

PARAMETER	DESCRIPTION
max_requests	Maximum number of requests allowed in the interval. TYPE: Union[int, Callable[..., int]]
interval	Time interval in seconds (default: 3600) TYPE: int DEFAULT: 3600
throttle_key	Custom key for the rate limit (default: function path) TYPE: Optional[str] DEFAULT: None
method	List of HTTP methods to apply limit to (optional) TYPE: Optional[list[str]] DEFAULT: None
scopes	List of user scopes to apply limit to (optional) TYPE: Optional[list[str]] DEFAULT: None

RETURNS	DESCRIPTION
Callable	Decorated function with rate limiting rules TYPE: Callable

Source code in webtool/throttle/decorator.py

def limiter(
    max_requests: Union[int, Callable[..., int]],
    interval: int = 3600,
    throttle_key: Optional[str] = None,
    method: Optional[list[str]] = None,
    scopes: Optional[list[str]] = None,
) -> Callable:
    """
    Decorator for implementing rate limiting on functions.

    Parameters:
        max_requests: Maximum number of requests allowed in the interval.
        interval: Time interval in seconds (default: 3600)
        throttle_key: Custom key for the rate limit (default: function path)
        method: List of HTTP methods to apply limit to (optional)
        scopes: List of user scopes to apply limit to (optional)

    Returns:
        Callable: Decorated function with rate limiting rules
    """

    def decorator(func):
        exist_func = _find_closure_rules_function(func)

        key = throttle_key
        if exist_func:
            key = key or sha256(f"{exist_func.__module__}{exist_func.__name__}{interval}{method}{scopes}").hex()
            if not hasattr(func, THROTTLE_RULE_ATTR_NAME):
                exist_rules = getattr(exist_func, THROTTLE_RULE_ATTR_NAME)
                setattr(func, THROTTLE_RULE_ATTR_NAME, exist_rules)
        else:
            key = key or sha256(f"{func.__module__}{func.__name__}{interval}{method}{scopes}").hex()
            setattr(func, THROTTLE_RULE_ATTR_NAME, LimitRuleManager())

        new_rule = LimitRule(
            max_requests=max_requests,
            interval=interval,
            throttle_key=key,
            method=[m.upper() for m in method] if method else [],
            scopes=scopes or [],
        )

        getattr(func, THROTTLE_RULE_ATTR_NAME).add_rules(new_rule)

        return func

    return decorator