Rate Keeper: Used to limit function call frequency. It ensures your function is called evenly within the limit rather than being called intensively in a short time. Moreover, it can dynamically adjust the call frequency based on remaining calls and time.
pip install rate-keeperfrom rate_keeper import RateKeeper
if __name__ == "__main__":
rate_keeper = RateKeeper(limit=3, period=1)
@rate_keeper.decorator
def request(url: str) -> str:
print(url, rate_keeper, f"{rate_keeper.delay_time:.2f}")
count = 0
while count < 6:
request(f"https://www.example.com/{count}")
count += 1
# Output:
# https://www.example.com/0 RateKeeper(limit=3, period=1, used=1, reset=89614.39) 0.00
# https://www.example.com/1 RateKeeper(limit=3, period=1, used=2, reset=89614.39) 0.50
# https://www.example.com/2 RateKeeper(limit=3, period=1, used=1, reset=89615.406) 0.48
# https://www.example.com/3 RateKeeper(limit=3, period=1, used=2, reset=89615.406) 0.50
# https://www.example.com/4 RateKeeper(limit=3, period=1, used=1, reset=89616.421) 0.49
# https://www.example.com/5 RateKeeper(limit=3, period=1, used=2, reset=89616.421) 0.50from datetime import datetime, timezone
from typing import Dict
import requests
from requests import Response
from rate_keeper import RateKeeper
# UTC timestamp clock
def timestamp_clock():
return datetime.now(timezone.utc).timestamp()
rate_keeper = RateKeeper(limit=5000, period=3600, clock=timestamp_clock)
@rate_keeper.decorator
def fetch(
method: str, url: str, headers: Dict[str, str] = {}, params: Dict[str, str] = {}
) -> Response:
# https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api#checking-the-status-of-your-rate-limit
response = requests.request(method, url, headers=headers, params=params)
headers_map = {
"x-ratelimit-limit": lambda x: setattr(rate_keeper, "limit", int(x)),
"x-ratelimit-used": lambda x: setattr(rate_keeper, "used", int(x)),
"x-ratelimit-reset": lambda x: setattr(rate_keeper, "reset", float(x)),
}
for key, value in response.headers.items():
lower_key = key.lower()
if lower_key in headers_map:
headers_map[lower_key](value)
return response
def create_headers(token: str) -> Dict[str, str]:
return {
"Accept": "application/vnd.github.v3+json",
"User-Agent": "Python requests GitHub API",
"Authorization": f"token {token}",
}
print(rate_keeper, f"{rate_keeper.recommend_delay:.2f}")
response = fetch("GET", "https://api.github.com/user", create_headers("github_token"))
print(response.json())
print(rate_keeper, f"{rate_keeper.recommend_delay:.2f}")
# Output:
# RateKeeper(limit=5000, period=3600, used=0, reset=1745897378.901664) 0.00
# {'message': 'Bad credentials', 'documentation_url': 'https://docs.github.com/rest', 'status': '401'}
# RateKeeper(limit=60, period=3600, used=3, reset=1745896988) 56.30