Skip to content

Rate Limiter

QPS and concurrency rate limiting with decorator support.

Rate limiting utilities with token-bucket and decorator-based APIs.

RateLimitException

Bases: Exception

Exception raised when a rate limit is exceeded.

Attributes:

Name Type Description
period_remaining

Time remaining until the rate limit resets.
Source code in shutils/rate_limiter.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class RateLimitException(Exception):  # noqa: N818
    """Exception raised when a rate limit is exceeded.

    Attributes:
        period_remaining: Time remaining until the rate limit resets.
    """

    def __init__(self, message, period_remaining):
        """Initialize the rate limit exception.

        Args:
            message: Exception message string.
            period_remaining: Seconds remaining until the rate limit resets.
        """
        super().__init__(message)
        self.period_remaining = period_remaining

__init__(message, period_remaining)

Initialize the rate limit exception.

Parameters:

Name Type Description Default
message

Exception message string.

 required
period_remaining

Seconds remaining until the rate limit resets.

 required
Source code in shutils/rate_limiter.py 
 96
 97
 98
 99
100
101
102
103
104
def __init__(self, message, period_remaining):
    """Initialize the rate limit exception.

    Args:
        message: Exception message string.
        period_remaining: Seconds remaining until the rate limit resets.
    """
    super().__init__(message)
    self.period_remaining = period_remaining

RateLimiter

A token-bucket rate limiter that tracks call counts within a time window.

Attributes:

Name Type Description
calls

Maximum number of calls allowed per period.
period

Time window in seconds.
Source code in shutils/rate_limiter.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class RateLimiter:
    """A token-bucket rate limiter that tracks call counts within a time window.

    Attributes:
        calls: Maximum number of calls allowed per period.
        period: Time window in seconds.
    """

    def __init__(self, calls: int, period: int = 1):
        """Initialize the rate limiter.

        Args:
            calls: Maximum number of allowed calls per period. Must be > 0.
            period: Time window in seconds. Must be >= 0.

        Raises:
            ValueError: If ``calls`` is <= 0 or ``period`` is < 0.
        """
        self.calls = calls
        self.period = period
        self.last_refill = self.now()
        self.lock = threading.Lock()
        self.tokens = 0
        if self.calls <= 0:
            raise ValueError("calls must be greater than 0")
        if self.period < 0:
            raise ValueError("period must be greater equal 0")

    @staticmethod
    def now():
        """Return a monotonic timestamp for the current time."""
        if hasattr(time, "monotonic"):
            return time.monotonic()
        else:
            return time.time()

    def __cal(self):
        if self.period == 0:
            return 0
        with self.lock:
            # 计算当前时间窗剩余时间
            current_time = self.now()
            time_since_last_refill = current_time - self.last_refill
            period_remaining = self.period - time_since_last_refill
            # 时间窗重制
            if period_remaining <= 0:
                self.tokens = 0
                self.last_refill = current_time

            self.tokens += 1

            # 判断是否超过限制
            if self.tokens > self.calls:
                return period_remaining

            return 0

    def allow(self):
        """Check whether a call is allowed within the current rate limit.

        Returns:
            True if the call is allowed, False otherwise.
        """
        return self.__cal() == 0

    def sleep_time(self):
        """Return the number of seconds to wait before the next call is allowed.

        Returns:
            Seconds remaining in the current rate-limit window, or 0 if allowed.
        """
        return self.__cal()

__init__(calls, period=1)

Initialize the rate limiter.

Parameters:

Name Type Description Default
calls int

Maximum number of allowed calls per period. Must be > 0.

 required
period int

Time window in seconds. Must be >= 0.

 1

Raises:

Type Description
ValueError

If calls is <= 0 or period is < 0.
Source code in shutils/rate_limiter.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, calls: int, period: int = 1):
    """Initialize the rate limiter.

    Args:
        calls: Maximum number of allowed calls per period. Must be > 0.
        period: Time window in seconds. Must be >= 0.

    Raises:
        ValueError: If ``calls`` is <= 0 or ``period`` is < 0.
    """
    self.calls = calls
    self.period = period
    self.last_refill = self.now()
    self.lock = threading.Lock()
    self.tokens = 0
    if self.calls <= 0:
        raise ValueError("calls must be greater than 0")
    if self.period < 0:
        raise ValueError("period must be greater equal 0")

allow()

Check whether a call is allowed within the current rate limit.

Returns:

Type Description

True if the call is allowed, False otherwise.
Source code in shutils/rate_limiter.py 
72
73
74
75
76
77
78
def allow(self):
    """Check whether a call is allowed within the current rate limit.

    Returns:
        True if the call is allowed, False otherwise.
    """
    return self.__cal() == 0

now() staticmethod

Return a monotonic timestamp for the current time.

Source code in shutils/rate_limiter.py 
43
44
45
46
47
48
49
@staticmethod
def now():
    """Return a monotonic timestamp for the current time."""
    if hasattr(time, "monotonic"):
        return time.monotonic()
    else:
        return time.time()

sleep_time()

Return the number of seconds to wait before the next call is allowed.

Returns:

Type Description

Seconds remaining in the current rate-limit window, or 0 if allowed.
Source code in shutils/rate_limiter.py 
80
81
82
83
84
85
86
def sleep_time(self):
    """Return the number of seconds to wait before the next call is allowed.

    Returns:
        Seconds remaining in the current rate-limit window, or 0 if allowed.
    """
    return self.__cal()

RateLimiterDecorator

Decorator for rate-limiting synchronous function calls.

Attributes:

Name Type Description
rate_limiter

The underlying RateLimiter instance.
sleep_and_retry

If True, sleep and retry instead of raising an exception.
raise_exception

If True, raise RateLimitException when rate-limited.
Source code in shutils/rate_limiter.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class RateLimiterDecorator:
    """Decorator for rate-limiting synchronous function calls.

    Attributes:
        rate_limiter: The underlying ``RateLimiter`` instance.
        sleep_and_retry: If True, sleep and retry instead of raising an exception.
        raise_exception: If True, raise ``RateLimitException`` when rate-limited.
    """

    def __init__(self, calls: int, period: int = 1, sleep_and_retry: bool = False, raise_exception: bool = True):
        """Initialize the rate limiter decorator.

        Args:
            calls: Maximum calls per period.
            period: Time window in seconds.
            sleep_and_retry: If True, sleep until the limit resets and retry.
            raise_exception: If True and not ``sleep_and_retry``, raise
                ``RateLimitException`` when the limit is exceeded.
        """
        self.rate_limiter = RateLimiter(calls, period)
        self.sleep_and_retry = sleep_and_retry
        self.raise_exception = raise_exception

    def __call__(self, func):
        """Wrap a function with rate-limiting logic.

        Args:
            func: The function to decorate.

        Returns:
            The rate-limited wrapper function.
        """
        @wraps(func)
        def wrapper(*args, **kwargs):
            while True:
                sleep_time = self.rate_limiter.sleep_time()
                if sleep_time == 0:
                    return func(*args, **kwargs)
                if self.sleep_and_retry:
                    time.sleep(sleep_time)
                else:
                    if self.raise_exception:
                        raise RateLimitException("Rate limit exceeded", sleep_time)
                    else:
                        return None
        return wrapper

__call__(func)

Wrap a function with rate-limiting logic.

Parameters:

Name Type Description Default
func

The function to decorate.

 required

Returns:

Type Description

The rate-limited wrapper function.
Source code in shutils/rate_limiter.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def __call__(self, func):
    """Wrap a function with rate-limiting logic.

    Args:
        func: The function to decorate.

    Returns:
        The rate-limited wrapper function.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        while True:
            sleep_time = self.rate_limiter.sleep_time()
            if sleep_time == 0:
                return func(*args, **kwargs)
            if self.sleep_and_retry:
                time.sleep(sleep_time)
            else:
                if self.raise_exception:
                    raise RateLimitException("Rate limit exceeded", sleep_time)
                else:
                    return None
    return wrapper

__init__(calls, period=1, sleep_and_retry=False, raise_exception=True)

Initialize the rate limiter decorator.

Parameters:

Name Type Description Default
calls int

Maximum calls per period.

 required
period int

Time window in seconds.

 1
sleep_and_retry bool

If True, sleep until the limit resets and retry.

 False
raise_exception bool

If True and not sleep_and_retry, raise RateLimitException when the limit is exceeded.

 True
Source code in shutils/rate_limiter.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
def __init__(self, calls: int, period: int = 1, sleep_and_retry: bool = False, raise_exception: bool = True):
    """Initialize the rate limiter decorator.

    Args:
        calls: Maximum calls per period.
        period: Time window in seconds.
        sleep_and_retry: If True, sleep until the limit resets and retry.
        raise_exception: If True and not ``sleep_and_retry``, raise
            ``RateLimitException`` when the limit is exceeded.
    """
    self.rate_limiter = RateLimiter(calls, period)
    self.sleep_and_retry = sleep_and_retry
    self.raise_exception = raise_exception