Metadata-Version: 2.3
Name: rate-keeper
Version: 0.2.1
Summary: 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.
Project-URL: Homepage, https://github.com/ftnfurina/rate-keeper#readme
Project-URL: Source, https://github.com/ftnfurina/rate-keeper
Project-URL: Tracker, https://github.com/ftnfurina/rate-keeper/issues
Author-email: ftnfurina <ftnfurina@gmail.com>
License: MIT
Keywords: api-rate-limiting,api-throttling,rate-limiter
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Utilities
Requires-Python: >=3.8
Requires-Dist: requests>=2.32.3
Description-Content-Type: text/markdown

![badge-collection](https://socialify.git.ci/ftnfurina/rate-keeper/image?font=Bitter&forks=1&issues=1&language=1&name=1&owner=1&pattern=Floating+Cogs&pulls=1&stargazers=1&theme=Auto)

<div align="center">
  <h1>Rate Keeper</h1>
  <p>
    <a href="https://github.com/ftnfurina/rate-keeper/blob/main/README_ZH.md">中文</a> |
    <a href="https://github.com/ftnfurina/rate-keeper/blob/main/README.md">English</a>
  </p>
</div>

**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.**

## Installation

```shell
pip install rate-keeper
```

## Quick Start

```python
from 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.50
```

## Dynamic Adjust

```python
from datetime import datetime, timezone
from typing import Dict

import requests
from requests import Response

from rate_keeper import RateKeeper

timestamp_clock = 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": rate_keeper.update_limit,
        "x-ratelimit-used": rate_keeper.update_used,
        "x-ratelimit-reset": rate_keeper.update_reset,
    }

    for key, value in response.headers.items():
        lower_key = key.lower()
        if lower_key in headers_map:
            headers_map[lower_key](int(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
```