API rate limit decorator for preventing function calls from exceeding specified frequency limits
npx @tessl/cli install tessl/pypi-ratelimit@2.2.0A Python decorator library for implementing API rate limiting functionality that prevents functions from being called more frequently than specified limits. It provides a clean, decorator-based approach to enforce rate limits on function calls, helping developers avoid hitting API rate limits that could result in service bans.
pip install ratelimitfrom ratelimit import limits, RateLimitException, sleep_and_retryAlternative imports:
# Using backwards-compatible alias
from ratelimit import rate_limited, RateLimitException, sleep_and_retry
# Package-level import
import ratelimit
# Access via: ratelimit.limits, ratelimit.RateLimitException, etc.from ratelimit import limits, RateLimitException, sleep_and_retry
import requests
# Basic rate limiting
FIFTEEN_MINUTES = 900
@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api(url):
response = requests.get(url)
if response.status_code != 200:
raise Exception('API response: {}'.format(response.status_code))
return response
# With exception handling
try:
result = call_api('https://api.example.com/data')
except RateLimitException as e:
print(f"Rate limit exceeded. Wait {e.period_remaining} seconds")
# Automatic retry on rate limit
@sleep_and_retry
@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api_with_retry(url):
response = requests.get(url)
if response.status_code != 200:
raise Exception('API response: {}'.format(response.status_code))
return responsePrevents functions from being called more frequently than specified limits. The main decorator that enforces rate limits on function calls.
class RateLimitDecorator(object):
def __init__(self, calls=15, period=900, clock=now, raise_on_limit=True):
"""
Rate limit decorator class. Available as 'limits' and 'rate_limited' aliases.
Parameters:
- calls (int): Maximum function invocations allowed within time period. Must be >= 1. Default: 15
- period (float): Time period in seconds before rate limit resets. Must be > 0. Default: 900 (15 minutes)
- clock (callable): Function returning current time as float. Default: time.monotonic or time.time
- raise_on_limit (bool): Whether to raise RateLimitException when limit exceeded. Default: True
"""
# Main alias (recommended)
limits = RateLimitDecorator
# Backwards compatibility alias
rate_limited = RateLimitDecoratorWraps rate-limited functions to automatically sleep and retry when rate limits are exceeded, ensuring every function call eventually succeeds.
def sleep_and_retry(func):
"""
Decorator that rescues rate limit exceptions by sleeping until the rate limit resets.
Parameters:
- func (callable): The function to decorate
Returns:
- callable: Decorated function that automatically retries on rate limit
"""Custom exception raised when the number of function invocations exceeds the imposed rate limit. Contains information about the remaining time until the rate limit resets.
class RateLimitException(Exception):
def __init__(self, message, period_remaining):
"""
Exception raised when rate limits are exceeded.
Parameters:
- message (str): Exception message
- period_remaining (float): Time remaining until rate limit resets
Attributes:
- period_remaining (float): Time in seconds until rate limit resets
"""Package version constant.
__version__ = "2.2.1"# Internal clock function used by the library (not exported)
# Uses monotonic time if available, otherwise falls back to system clock
now = time.monotonic if hasattr(time, 'monotonic') else time.timefrom ratelimit import limits
import time
class TestClock:
def __init__(self):
self.time = 0
def __call__(self):
return self.time
def advance(self, seconds):
self.time += seconds
clock = TestClock()
@limits(calls=2, period=10, clock=clock)
def test_function():
return "success"
# Function can be called twice
test_function() # Success
test_function() # Success
# Third call raises exception
try:
test_function() # RateLimitException
except RateLimitException:
print("Rate limit exceeded")
# Advance clock and try again
clock.advance(10)
test_function() # Success againfrom ratelimit import limits
@limits(calls=1, period=60, raise_on_limit=False)
def api_call():
print("API called")
return "data"
# First call succeeds
result = api_call() # Prints "API called", returns "data"
# Second call silently skips execution
result = api_call() # Returns None, no print statementfrom ratelimit import limits, RateLimitException, sleep_and_retry
from backoff import on_exception, expo
@on_exception(expo, RateLimitException, max_tries=8)
@limits(calls=15, period=900)
def api_call_with_backoff(url):
# Your API call implementation
pass
# Alternative using sleep_and_retry (simpler but less flexible)
@sleep_and_retry
@limits(calls=15, period=900)
def api_call_simple_retry(url):
# Your API call implementation
passAll decorators are thread-safe using threading.RLock(). Each decorated function maintains its own independent rate limit state, allowing multiple rate-limited functions to operate concurrently without interference.
The package raises RateLimitException when rate limits are exceeded (unless raise_on_limit=False). This exception contains:
period_remaining attribute: float indicating seconds until rate limit resetsCommon error handling patterns:
import time
from ratelimit import limits, RateLimitException
@limits(calls=1, period=60)
def rate_limited_function():
return "success"
# Manual retry with exponential backoff
max_retries = 3
for attempt in range(max_retries):
try:
result = rate_limited_function()
break
except RateLimitException as e:
if attempt == max_retries - 1:
raise
wait_time = min(e.period_remaining * (2 ** attempt), 300) # Cap at 5 minutes
time.sleep(wait_time)