# Ratehawk
Ratehawk is a flexible API rate limiting library for Python. It allows you to easily implement rate limiting in your applications to prevent abuse and ensure fair usage of your APIs.
## Installation
You can install Ratehawk using pip:
```sh
pip install ratehawk
```
## Usage
Here are some examples of how to use Ratehawk in your Python applications:
### Basic Usage
```python
from ratehawk import RateLimiter
# Create a rate limiter with a limit of 10 requests per minute
rate_limiter = RateLimiter(limits=[(10, 60)])
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Redis Storage
```python
from ratehawk import RateLimiter
from ratehawk.storage.redis import RedisStorage
from redis import Redis
# Create a Redis client
redis_client = Redis(host='localhost', port=6379, db=0)
# Create a rate limiter with Redis storage
rate_limiter = RateLimiter(limits=[(10, 60)], storage=RedisStorage(redis_client))
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Postgres Storage
```python
from ratehawk import RateLimiter
from ratehawk.storage.postgres import PostgresStorage
# Create a rate limiter with Postgres storage
rate_limiter = RateLimiter(limits=[(10, 60)], storage=PostgresStorage(dsn="postgresql://localhost/ratehawk"))
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using SQLite Storage
```python
from ratehawk import RateLimiter
from ratehawk.storage.sqlite import SQLiteStorage
# Create a rate limiter with SQLite storage
rate_limiter = RateLimiter(limits=[(10, 60)], storage=SQLiteStorage(db_path="ratehawk.db"))
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using RateLimitMetrics
```python
from ratehawk import RateLimiter
from ratehawk.monitoring.metrics import RateLimitMetrics
# Create a rate limiter with metrics monitoring
rate_limiter = RateLimiter(limits=[(10, 60)])
rate_limiter.metrics = RateLimitMetrics()
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using RateLimitLogger
```python
from ratehawk import RateLimiter
from ratehawk.monitoring.logging import RateLimitLogger
# Create a rate limiter with logging
rate_limiter = RateLimiter(limits=[(10, 60)])
rate_limiter.logger = RateLimitLogger()
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using RateLimitEvent
```python
from ratehawk import RateLimiter
from ratehawk.monitoring.events import RateLimitEvent, EventEmitter
# Create a rate limiter with event monitoring
rate_limiter = RateLimiter(limits=[(10, 60)])
rate_limiter.events = EventEmitter()
# Register event handlers
def on_limit_exceeded(key, retry_after):
print(f"Rate limit exceeded for key: {key}, retry after: {retry_after}s")
def on_near_limit(key, current, limit):
print(f"Near rate limit for key: {key}, current: {current}, limit: {limit}")
def on_reset(key):
print(f"Rate limit reset for key: {key}")
rate_limiter.events.on(RateLimitEvent.LIMIT_EXCEEDED, on_limit_exceeded)
rate_limiter.events.on(RateLimitEvent.NEAR_LIMIT, on_near_limit)
rate_limiter.events.on(RateLimitEvent.RESET, on_reset)
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Multiple Rate Limits per Key
```python
from ratehawk import RateLimiter
# Create a rate limiter with multiple rate limits per key
rate_limiter = RateLimiter(limits=[(10, 60), (100, 3600)])
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Rate Limit Groups and Hierarchies
```python
from ratehawk import RateLimiter
# Create a rate limiter with rate limit groups and hierarchies
rate_limiter = RateLimiter(
limits=[(10, 60)],
burst_limits=[(20, 15)],
quota_limits=[(1000, 86400)]
)
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Dynamic Rate Limits
```python
from ratehawk import RateLimiter
# Function to dynamically determine rate limits based on user attributes or time of day
def dynamic_limits_func():
# Example: Different rate limits for different user roles
user_role = get_user_role()
if user_role == "premium":
return {"limits": [(100, 60)], "burst_limits": [(200, 15)], "quota_limits": [(5000, 86400)]}
else:
return {"limits": [(10, 60)], "burst_limits": [(20, 15)], "quota_limits": [(1000, 86400)]}
# Create a rate limiter with dynamic rate limits
rate_limiter = RateLimiter(
limits=[(10, 60)],
dynamic_limits_func=dynamic_limits_func
)
# Apply dynamic rate limits
await rate_limiter.apply_dynamic_limits()
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
```
### Using Rate Limit Quotas
```python
from ratehawk import RateLimiter
# Create a rate limiter with rate limit quotas over longer periods
rate_limiter = RateLimiter(
limits=[(10, 60)],
quota_limits=[(1000, 86400)]
)
# Check if the rate limit is exceeded
if await rate_limiter.check():
print("Rate limit not exceeded")
else:
print("Rate limit exceeded")
# Increment the rate limit counter
try:
await rate_limiter.increment()
print("Request allowed")
except RateLimitExceeded:
print("Rate limit exceeded, try again later")
# Check if the quota limit is exceeded
if await rate_limiter.check_quota():
print("Quota limit not exceeded")
else:
print("Quota limit exceeded")
# Increment the quota limit counter
try:
await rate_limiter.increment_quota()
print("Request allowed")
except RateLimitExceeded:
print("Quota limit exceeded, try again later")
```
## Examples
For more detailed examples, please refer to the following files in the `docs` folder:
- [Basic Example](docs/basic_example.md)
- [FastAPI Example](docs/fastapi_example.md)
- [Flask Example](docs/flask_example.md)
## License
This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more details.
Raw data
{
"_id": null,
"home_page": "https://github.com/wadedesign/ratehawk",
"name": "ratehawk",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": null,
"author": "Andrew Wade",
"author_email": "hi@wadedev.online",
"download_url": "https://files.pythonhosted.org/packages/db/e4/7633cdd0e62dd8c19e8d3141c5503a766ac315934cdb515556b608e83a8b/ratehawk-0.2.1.tar.gz",
"platform": null,
"description": "# Ratehawk\n\nRatehawk is a flexible API rate limiting library for Python. It allows you to easily implement rate limiting in your applications to prevent abuse and ensure fair usage of your APIs.\n\n## Installation\n\nYou can install Ratehawk using pip:\n\n```sh\npip install ratehawk\n```\n\n## Usage\n\nHere are some examples of how to use Ratehawk in your Python applications:\n\n### Basic Usage\n\n```python\nfrom ratehawk import RateLimiter\n\n# Create a rate limiter with a limit of 10 requests per minute\nrate_limiter = RateLimiter(limits=[(10, 60)])\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Redis Storage\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.storage.redis import RedisStorage\nfrom redis import Redis\n\n# Create a Redis client\nredis_client = Redis(host='localhost', port=6379, db=0)\n\n# Create a rate limiter with Redis storage\nrate_limiter = RateLimiter(limits=[(10, 60)], storage=RedisStorage(redis_client))\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Postgres Storage\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.storage.postgres import PostgresStorage\n\n# Create a rate limiter with Postgres storage\nrate_limiter = RateLimiter(limits=[(10, 60)], storage=PostgresStorage(dsn=\"postgresql://localhost/ratehawk\"))\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using SQLite Storage\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.storage.sqlite import SQLiteStorage\n\n# Create a rate limiter with SQLite storage\nrate_limiter = RateLimiter(limits=[(10, 60)], storage=SQLiteStorage(db_path=\"ratehawk.db\"))\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using RateLimitMetrics\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.monitoring.metrics import RateLimitMetrics\n\n# Create a rate limiter with metrics monitoring\nrate_limiter = RateLimiter(limits=[(10, 60)])\nrate_limiter.metrics = RateLimitMetrics()\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using RateLimitLogger\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.monitoring.logging import RateLimitLogger\n\n# Create a rate limiter with logging\nrate_limiter = RateLimiter(limits=[(10, 60)])\nrate_limiter.logger = RateLimitLogger()\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using RateLimitEvent\n\n```python\nfrom ratehawk import RateLimiter\nfrom ratehawk.monitoring.events import RateLimitEvent, EventEmitter\n\n# Create a rate limiter with event monitoring\nrate_limiter = RateLimiter(limits=[(10, 60)])\nrate_limiter.events = EventEmitter()\n\n# Register event handlers\ndef on_limit_exceeded(key, retry_after):\n print(f\"Rate limit exceeded for key: {key}, retry after: {retry_after}s\")\n\ndef on_near_limit(key, current, limit):\n print(f\"Near rate limit for key: {key}, current: {current}, limit: {limit}\")\n\ndef on_reset(key):\n print(f\"Rate limit reset for key: {key}\")\n\nrate_limiter.events.on(RateLimitEvent.LIMIT_EXCEEDED, on_limit_exceeded)\nrate_limiter.events.on(RateLimitEvent.NEAR_LIMIT, on_near_limit)\nrate_limiter.events.on(RateLimitEvent.RESET, on_reset)\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Multiple Rate Limits per Key\n\n```python\nfrom ratehawk import RateLimiter\n\n# Create a rate limiter with multiple rate limits per key\nrate_limiter = RateLimiter(limits=[(10, 60), (100, 3600)])\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Rate Limit Groups and Hierarchies\n\n```python\nfrom ratehawk import RateLimiter\n\n# Create a rate limiter with rate limit groups and hierarchies\nrate_limiter = RateLimiter(\n limits=[(10, 60)],\n burst_limits=[(20, 15)],\n quota_limits=[(1000, 86400)]\n)\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Dynamic Rate Limits\n\n```python\nfrom ratehawk import RateLimiter\n\n# Function to dynamically determine rate limits based on user attributes or time of day\ndef dynamic_limits_func():\n # Example: Different rate limits for different user roles\n user_role = get_user_role()\n if user_role == \"premium\":\n return {\"limits\": [(100, 60)], \"burst_limits\": [(200, 15)], \"quota_limits\": [(5000, 86400)]}\n else:\n return {\"limits\": [(10, 60)], \"burst_limits\": [(20, 15)], \"quota_limits\": [(1000, 86400)]}\n\n# Create a rate limiter with dynamic rate limits\nrate_limiter = RateLimiter(\n limits=[(10, 60)],\n dynamic_limits_func=dynamic_limits_func\n)\n\n# Apply dynamic rate limits\nawait rate_limiter.apply_dynamic_limits()\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n```\n\n### Using Rate Limit Quotas\n\n```python\nfrom ratehawk import RateLimiter\n\n# Create a rate limiter with rate limit quotas over longer periods\nrate_limiter = RateLimiter(\n limits=[(10, 60)],\n quota_limits=[(1000, 86400)]\n)\n\n# Check if the rate limit is exceeded\nif await rate_limiter.check():\n print(\"Rate limit not exceeded\")\nelse:\n print(\"Rate limit exceeded\")\n\n# Increment the rate limit counter\ntry:\n await rate_limiter.increment()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Rate limit exceeded, try again later\")\n\n# Check if the quota limit is exceeded\nif await rate_limiter.check_quota():\n print(\"Quota limit not exceeded\")\nelse:\n print(\"Quota limit exceeded\")\n\n# Increment the quota limit counter\ntry:\n await rate_limiter.increment_quota()\n print(\"Request allowed\")\nexcept RateLimitExceeded:\n print(\"Quota limit exceeded, try again later\")\n```\n\n## Examples\n\nFor more detailed examples, please refer to the following files in the `docs` folder:\n\n- [Basic Example](docs/basic_example.md)\n- [FastAPI Example](docs/fastapi_example.md)\n- [Flask Example](docs/flask_example.md)\n\n## License\n\nThis project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more details.\n",
"bugtrack_url": null,
"license": null,
"summary": "A flexible API rate limiting library for Python",
"version": "0.2.1",
"project_urls": {
"Homepage": "https://github.com/wadedesign/ratehawk"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "c9a00078e0462d3af68053352f41ea4c43c11e7c50a4673debb0499e03f2cfc0",
"md5": "3070b07f6e60845fba7882cbcb9241ba",
"sha256": "e21f8922530f846d03a8db28d706908cc5b1a948726b76fd49ed629a041a78e9"
},
"downloads": -1,
"filename": "ratehawk-0.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3070b07f6e60845fba7882cbcb9241ba",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 16784,
"upload_time": "2024-11-09T09:38:56",
"upload_time_iso_8601": "2024-11-09T09:38:56.508731Z",
"url": "https://files.pythonhosted.org/packages/c9/a0/0078e0462d3af68053352f41ea4c43c11e7c50a4673debb0499e03f2cfc0/ratehawk-0.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "dbe47633cdd0e62dd8c19e8d3141c5503a766ac315934cdb515556b608e83a8b",
"md5": "67ac6fa1453d99ecddf3065b79dfd047",
"sha256": "0c832c482e220a5eeab0eeb3e8ab161bf5b5249a068a8429ddb63096ad0c6e95"
},
"downloads": -1,
"filename": "ratehawk-0.2.1.tar.gz",
"has_sig": false,
"md5_digest": "67ac6fa1453d99ecddf3065b79dfd047",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 16596,
"upload_time": "2024-11-09T09:38:58",
"upload_time_iso_8601": "2024-11-09T09:38:58.145452Z",
"url": "https://files.pythonhosted.org/packages/db/e4/7633cdd0e62dd8c19e8d3141c5503a766ac315934cdb515556b608e83a8b/ratehawk-0.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-09 09:38:58",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "wadedesign",
"github_project": "ratehawk",
"github_not_found": true,
"lcname": "ratehawk"
}
JSON
