<img align="left" width="95" height="120" src="https://raw.githubusercontent.com/vutran1710/PyrateLimiter/master/docs/_static/logo.png">
# PyrateLimiter
The request rate limiter using Leaky-bucket Algorithm.
Full project documentation can be found at [pyratelimiter.readthedocs.io](https://pyratelimiter.readthedocs.io).
[](https://badge.fury.io/py/pyrate-limiter)
[](https://pypi.org/project/pyrate-limiter)
[](https://codecov.io/gh/vutran1710/PyrateLimiter)
[](https://github.com/vutran1710/PyrateLimiter/graphs/commit-activity)
[](https://pypi.python.org/pypi/pyrate-limiter/)
<br>
## Contents
- [PyrateLimiter](#pyratelimiter)
- [Contents](#contents)
- [Features](#features)
- [Installation](#installation)
- [Quickstart](#quickstart)
- [Basic usage](#basic-usage)
- [Key concepts](#key-concepts)
- [Defining rate limits & buckets](#defining-rate-limits-and-buckets)
- [Defining clock & routing logic](#defining-clock--routing-logic-with-bucketfactory)
- [Wrapping all up with Limiter](#wrapping-all-up-with-limiter)
- [Weight](#weight)
- [Handling exceeded limits](#handling-exceeded-limits)
- [Bucket analogy](#bucket-analogy)
- [Rate limit exceptions](#rate-limit-exceptions)
- [Rate limit delays](#rate-limit-delays)
- [Backends](#backends)
- [InMemoryBucket](#inmemorybucket)
- [SQLiteBucket](#sqlitebucket)
- [RedisBucket](#redisbucket)
- [PostgresBucket](#postgresbucket)
- [Decorator](#decorator)
- [Advanced Usage](#advanced-usage)
- [Component-level Diagram](#component-level-diagram)
- [Time sources](#time-sources)
- [Leaking](#leaking)
- [Concurrency](#concurrency)
- [Custom backend](#custom-backend)
## Features
- Tracks any number of rate limits and intervals you want to define
- Independently tracks rate limits for multiple services or resources
- Handles exceeded rate limits by either raising errors or adding delays
- Several usage options including a normal function call, a decorator
- Out-of-the-box workable with both sync & async
- Includes optional SQLite and Redis backends, which can be used to persist limit tracking across
multiple threads, or application restarts
## Installation
**PyrateLimiter** supports **python ^3.8**
Install using pip:
```
pip install pyrate-limiter
```
Or using conda:
```
conda install --channel conda-forge pyrate-limiter
```
## Quickstart
Let's say you want to limit 5 requests over 2 seconds, and raise an exception if the limit is exceeded:
```python
from pyrate_limiter import Duration, Rate, Limiter, BucketFullException
rate = Rate(5, Duration.SECOND * 2)
limiter = Limiter(rate)
# Or you can pass multiple rates
# rates = [Rate(5, Duration.SECOND * 2), Rate(10, Duration.MINUTE)]
# limiter = Limiter(rates)
for request in range(6):
try:
limiter.try_acquire(request)
except BucketFullException as err:
print(err)
print(err.meta_info)
# Bucket for item=5 with Rate limit=5/2.0s is already full
# {'error': 'Bucket for item=5 with Rate limit=5/2.0s is already full', 'name': 5, 'weight': 1, 'rate': 'limit=5/2.0s'}
```
## Basic Usage
### Key concepts
#### Clock
- Timestamp items
#### Bucket
- Hold items with timestamps.
- Behave like a FIFO queue
- It can `leak` - popping items that are no longer relevant out of the queue
#### BucketFactory
- BucketFactory keeps references to buckets & clocks: determine the exact time that items arrive then route them to their corresponding buckets
- Help schedule background tasks to run buckets' `leak` periodically to make sure buckets will not explode from containing too many items
- Where user define his own logic: routing, condition-checking, timing etc...
#### Limiter
The Limiter's most important responsibility is to make user's life as easiest as possible:
- Sums up all the underlying logic to a simple, intuitive API to work with
- Handles async/sync context seamlessly (everything just `works` by adding/removing `async/await` keyword to the user's code)
- Provides different ways of interacting with the underlying **BucketFactory** _(plain method call, decorator, context-manager (TBA))_
- Provides thread-safety using RLock
### Defining rate limits and buckets
Consider some public API (like LinkedIn, GitHub, etc.) that has rate limits like the following:
```
- 500 requests per hour
- 1000 requests per day
- 10000 requests per month
```
You can define these rates using the `Rate` class. `Rate` class has 2 properties only: **limit** and **interval**
```python
from pyrate_limiter import Duration, Rate
hourly_rate = Rate(500, Duration.HOUR) # 500 requests per hour
daily_rate = Rate(1000, Duration.DAY) # 1000 requests per day
monthly_rate = Rate(10000, Duration.WEEK * 4) # 10000 requests per month
rates = [hourly_rate, daily_rate, monthly_rate]
```
Rates must be properly ordered:
- Rates' intervals & limits must be ordered from least to greatest
- Rates' ratio of **limit/interval** must be ordered from greatest to least
Existing implementations of Bucket come with rate-validation when init. If you are to use your own implementation, use the validator provided by the lib
```python
from pyrate_limiter import validate_rate_list
assert validate_rate_list(my_rates)
```
Then, add the rates to the bucket of your choices
```python
from pyrate_limiter import InMemoryBucket, RedisBucket
basic_bucket = InMemoryBucket(rates)
# Or, using redis
from redis import Redis
redis_connection = Redis(host='localhost')
redis_bucket = RedisBucket.init(rates, redis_connection, "my-bucket-name")
# Async Redis would work too!
from redis.asyncio import Redis
redis_connection = Redis(host='localhost')
redis_bucket = await RedisBucket.init(rates, redis_connection, "my-bucket-name")
```
If you only need a single Bucket for everything, and python's built-in `time()` is enough for you, then pass the bucket to Limiter then ready to roll!
```python
from pyrate_limiter import Limiter
# Limiter constructor accepts single bucket as the only parameter,
# the rest are 3 optional parameters with default values as following
# Limiter(bucket, clock=TimeClock(), raise_when_fail=True, max_delay=None)
limiter = Limiter(bucket)
# Limiter is now ready to work!
limiter.try_acquire("hello world")
```
If you want to have finer grain control with routing & clocks etc, then you should use `BucketFactory`.
### Defining Clock & routing logic with BucketFactory
If you need more than one type of Bucket, and be able to route items to different buckets based on some condition, you can use BucketFactory to do that.
From the above steps, you already have your buckets. Now it's time to define what `Time` is (funny?!). Most of the time (again!?), you can use the existing Clock backend provided by **pyrate_limiter**.
```python
from pyrate_limiter.clock import TimeClock, MonotonicClock, SQLiteClock
base_clock = TimeClock()
```
**PyrateLimiter** makes no assumption about users logic, so to map coming items to their correct buckets, implement your own **BucketFactory** class! At minimum, there are only 2 methods require implementing
```python
from pyrate_limiter import BucketFactory
from pyrate_limiter import AbstractBucket
class MyBucketFactory(BucketFactory):
# You can use constructor here,
# nor it requires to make bucket-factory work!
def wrap_item(self, name: str, weight: int = 1) -> RateItem:
"""Time-stamping item, return a RateItem"""
now = clock.now()
return RateItem(name, now, weight=weight)
def get(self, _item: RateItem) -> AbstractBucket:
"""For simplicity's sake, all items route to the same, single bucket"""
return bucket
```
### Creating buckets dynamically
If more than one bucket is needed, the bucket-routing logic should go to BucketFactory `get(..)` method.
When creating buckets dynamically, it is needed to schedule leak for each newly created buckets.
To support this, BucketFactory comes with a predefined method call `self.create(..)`. It is meant to create the bucket and schedule that bucket for leaking using the Factory's clock
```python
def create(
self,
clock: AbstractClock,
bucket_class: Type[AbstractBucket],
*args,
**kwargs,
) -> AbstractBucket:
"""Creating a bucket dynamically"""
bucket = bucket_class(*args, **kwargs)
self.schedule_leak(bucket, clock)
return bucket
```
By utilizing this, we can modify the code as following:
```python
class MultiBucketFactory(BucketFactory):
def __init__(self, clock):
self.clock = clock
self.buckets = {}
def wrap_item(self, name: str, weight: int = 1) -> RateItem:
"""Time-stamping item, return a RateItem"""
now = clock.now()
return RateItem(name, now, weight=weight)
def get(self, item: RateItem) -> AbstractBucket:
if item.name not in self.buckets:
# Use `self.create(..)` method to both initialize new bucket and calling `schedule_leak` on that bucket
# We can create different buckets with different types/classes here as well
new_bucket = self.create(YourBucketClass, *your-arguments, **your-keyword-arguments)
self.buckets.update({item.name: new_bucket})
return self.buckets[item.name]
```
### Wrapping all up with Limiter
Pass your bucket-factory to Limiter, and ready to roll!
```python
from pyrate_limiter import Limiter
limiter = Limiter(
bucket_factory,
raise_when_fail=False, # Default = True
max_delay=1000, # Default = None
)
item = "the-earth"
limiter.try_acquire(item)
heavy_item = "the-sun"
limiter.try_acquire(heavy_item, weight=10000)
```
If your bucket's backend is `async`, well, we got you covered! Passing `await` to the limiter is enought to make it scream!
```python
await limiter.try_acquire(item)
```
Alternatively, you can use `Limiter.try_acquire` as a function decorator. But you have to provide a `mapping` function that map the wrapped function's arguments to a proper `limiter.try_acquire` argument - which is a tuple of `(str, int)` or just `str`
```python
my_beautiful_decorator = limiter.as_decorator()
def mapping(some_number: int):
return str(some_number)
@my_beautiful_decorator(mapping)
def request_function(some_number: int):
requests.get('https://example.com')
# Async would work too!
@my_beautiful_decorator(mapping)
async def async_request_function(some_number: int):
requests.get('https://example.com')
```
### Weight
Item can have weight. By default item's weight = 1, but you can modify the weight before passing to `limiter.try_acquire`.
Item with weight W > 1 when consumed will be multiplied to (W) items with the same timestamp and weight = 1. Example with a big item with weight W=5, when put to bucket, it will be divided to 5 items with weight=1 + following names
```
BigItem(weight=5, name="item", timestamp=100) => [
item(weight=1, name="item", timestamp=100),
item(weight=1, name="item", timestamp=100),
item(weight=1, name="item", timestamp=100),
item(weight=1, name="item", timestamp=100),
item(weight=1, name="item", timestamp=100),
]
```
Yet, putting this big, heavy item into bucket is expected to be transactional & atomic - meaning either all 5 items will be consumed or none of them will. This is made possible as bucket `put(item)` always check for available space before ingesting. All of the Bucket's implementations provided by **PyrateLimiter** follows this rule.
Any additional, custom implementation of Bucket are expected to behave alike - as we have unit tests to cover the case.
See [Advanced usage options](#advanced-usage) below for more details.
### Handling exceeded limits
When a rate limit is exceeded, you have two options: raise an exception, or add delays.
#### Bucket analogy
<img height="300" align="right" src="https://upload.wikimedia.org/wikipedia/commons/c/c4/Leaky_bucket_analogy.JPG">
At this point it's useful to introduce the analogy of "buckets" used for rate-limiting. Here is a
quick summary:
- This library implements the [Leaky Bucket algorithm](https://en.wikipedia.org/wiki/Leaky_bucket).
- It is named after the idea of representing some kind of fixed capacity -- like a network or service -- as a bucket.
- The bucket "leaks" at a constant rate. For web services, this represents the **ideal or permitted request rate**.
- The bucket is "filled" at an intermittent, unpredicatble rate, representing the **actual rate of requests**.
- When the bucket is "full", it will overflow, representing **canceled or delayed requests**.
- Item can have weight. Consuming a single item with weight W > 1 is the same as consuming W smaller, unit items - each with weight=1, with the same timestamp and maybe same name (depending on however user choose to implement it)
#### Rate limit exceptions
By default, a `BucketFullException` will be raised when a rate limit is exceeded.
The error contains a `meta_info` attribute with the following information:
- `name`: The name of item it received
- `weight`: The weight of item it received
- `rate`: The specific rate that has been exceeded
Here's an example that will raise an exception on the 4th request:
```python
rate = Rate(3, Duration.SECOND)
bucket = InMemoryBucket([rate])
clock = TimeClock()
class MyBucketFactory(BucketFactory):
def wrap_item(self, name: str, weight: int = 1) -> RateItem:
"""Time-stamping item, return a RateItem"""
now = clock.now()
return RateItem(name, now, weight=weight)
def get(self, _item: RateItem) -> AbstractBucket:
"""For simplicity's sake, all items route to the same, single bucket"""
return bucket
limiter = Limiter(MyBucketFactory())
for _ in range(4):
try:
limiter.try_acquire('item', weight=2)
except BucketFullException as err:
print(err)
# Output: Bucket with Rate 3/1.0s is already full
print(err.meta_info)
# Output: {'name': 'item', 'weight': 2, 'rate': '3/1.0s', 'error': 'Bucket with Rate 3/1.0s is already full'}
```
The rate part of the output is constructed as: `limit / interval`. On the above example, the limit
is 3 and the interval is 1, hence the `Rate 3/1`.
#### Rate limit delays
You may want to simply slow down your requests to stay within the rate limits instead of canceling
them. In that case you pass the `max_delay` argument the maximum value of delay (typically in _ms_ when use human-clock).
```python
limiter = Limiter(factory, max_delay=500) # Allow to delay up to 500ms
```
As `max_delay` has been passed as a numeric value, when ingesting item, limiter will:
- First, try to ingest such item using the routed bucket
- If it fails to put item into the bucket, it will call `wait(item)` on the bucket to see how much time remains until the bucket can consume the item again?
- Comparing the `wait` value to the `max_delay`.
- if `max_delay` >= `wait`: delay (wait + 50ms as latency-tolerance) using either `asyncio.sleep` or `time.sleep` until the bucket can consume again
- if `max_delay` < `wait`: it raises `LimiterDelayException` if Limiter's `raise_when_fail=True`, otherwise silently fail and return False
Example:
```python
from pyrate_limiter import LimiterDelayException
for _ in range(4):
try:
limiter.try_acquire('item', weight=2, max_delay=200)
except LimiterDelayException as err:
print(err)
# Output:
# Actual delay exceeded allowance: actual=500, allowed=200
# Bucket for 'item' with Rate 3/1.0s is already full
print(err.meta_info)
# Output: {'name': 'item', 'weight': 2, 'rate': '3/1.0s', 'max_delay': 200, 'actual_delay': 500}
```
### Backends
A few different bucket backends are available:
- **InMemoryBucket**: using python built-in list as bucket
- **RedisBucket**, using err... redis, with both async/sync support
- **PostgresBucket**, using `psycopg2`
- **SQLiteBucket**, using sqlite3
#### InMemoryBucket
The default bucket is stored in memory, using python `list`
```python
from pyrate_limiter import InMemoryBucket, Rate, Duration
rates = [Rate(5, Duration.MINUTE * 2)]
bucket = InMemoryBucket(rates)
```
This bucket only availabe in `sync` mode. The only constructor argument is `List[Rate]`.
#### RedisBucket
RedisBucket uses `Sorted-Set` to store items with key being item's name and score item's timestamp
Because it is intended to work with both async & sync, we provide a classmethod `init` for it
```python
from pyrate_limiter import RedisBucket, Rate, Duration
# Using synchronous redis
from redis import ConnectionPool
from redis import Redis
rates = [Rate(5, Duration.MINUTE * 2)]
pool = ConnectionPool.from_url("redis://localhost:6379")
redis_db = Redis(connection_pool=pool)
bucket_key = "bucket-key"
bucket = RedisBucket.init(rates, redis_db, bucket_key)
# Using asynchronous redis
from redis.asyncio import ConnectionPool as AsyncConnectionPool
from redis.asyncio import Redis as AsyncRedis
pool = AsyncConnectionPool.from_url("redis://localhost:6379")
redis_db = AsyncRedis(connection_pool=pool)
bucket_key = "bucket-key"
bucket = await RedisBucket.init(rates, redis_db, bucket_key)
```
The API are the same, regardless of sync/async. If AsyncRedis is being used, calling `await bucket.method_name(args)` would just work!
#### SQLiteBucket
If you need to persist the bucket state, a SQLite backend is available.
Manully create a connection to Sqlite and pass it along with the table name to the bucket class:
```python
from pyrate_limiter import SQLiteBucket, Rate, Duration
import sqlite3
rates = [Rate(5, Duration.MINUTE * 2)]
conn = sqlite3.connect(
"/var/mydb.sqlite",
isolation_level="EXCLUSIVE",
check_same_thread=False,
)
table = "my-bucket-table"
bucket = SQLiteBucket(rates, conn, table)
```
#### PostgresBucket
Postgres is supported, but you have to install `psycopg[pool]` either as an extra or as a separate package.
You can use Postgres's built-in **CURRENT_TIMESTAMP** as the time source with `PostgresClock`, or use an external custom time source.
```python
from pyrate_limiter import PostgresBucket, Rate, PostgresClock
from psycopg_pool import ConnectionPool
connection_pool = ConnectionPool('postgresql://postgres:postgres@localhost:5432')
clock = PostgresClock(connection_pool)
rates = [Rate(3, 1000), Rate(4, 1500)]
bucket = PostgresBucket(connection_pool, "my-bucket-table", rates)
```
### Decorator
Limiter can be used as decorator, but you have to provide a `mapping` function that maps the wrapped function's arguments to `limiter.try_acquire` function arguments. The mapping function must return either a tuple of `(str, int)` or just a `str`
The decorator can work with both sync & async function
```python
decorator = limiter.as_decorator()
def mapping(*args, **kwargs):
return "demo", 1
@decorator(mapping)
def handle_something(*args, **kwargs):
"""function logic"""
@decorator(mapping)
async def handle_something_async(*args, **kwargs):
"""function logic"""
```
## Advanced Usage
### Component level diagram
### Time sources
Time source can be anything from anywhere: be it python's built-in time, or monotonic clock, sqliteclock, or crawling from world time server(well we don't have that, but you can!).
```python
from pyrate_limiter import TimeClock # use python' time.time()
from pyrate_limiter import MonotonicClock # use python time.monotonic()
```
Clock's abstract interface only requires implementing a method `now() -> int`. And it can be both sync or async.
### Leaking
Typically bucket should not hold items forever. Bucket's abstract interface requires its implementation must be provided with `leak(current_timestamp: Optional[int] = None)`.
The `leak` method when called is expected to remove any items considered outdated at that moment. During Limiter lifetime, all the buckets' `leak` should be called periodically.
**BucketFactory** provide a method called `schedule_leak` to help deal with this matter. Basically, it will run as a background task for all the buckets currently in use, with interval between `leak` call by **default is 10 seconds**.
```python
# Runnning a background task (whether it is sync/async - doesnt matter)
# calling the bucket's leak
factory.schedule_leak(bucket, clock)
```
You can change this calling interval by overriding BucketFactory's `leak_interval` property. This interval is in **miliseconds**.
```python
class MyBucketFactory(BucketFactory):
def __init__(self, *args):
self.leak_interval = 300
```
When dealing with leak using BucketFactory, the author's suggestion is, we can be pythonic about this by implementing a constructor
```python
class MyBucketFactory(BucketFactory):
def constructor(self, clock, buckets):
self.clock = clock
self.buckets = buckets
for bucket in buckets:
self.schedule_leak(bucket, clock)
```
### Concurrency
Generally, Lock is provided at Limiter's level, except SQLiteBucket case.
### Custom backends
If these don't suit your needs, you can also create your own bucket backend by implementing `pyrate_limiter.AbstractBucket` class.
One of **PyrateLimiter** design goals is powerful extensibility and maximum ease of development.
It must be not only be a ready-to-use tool, but also a guide-line, or a framework that help implementing new features/bucket free of the most hassles.
Due to the composition nature of the library, it is possbile to write minimum code and validate the result:
- Fork the repo
- Implement your bucket with `pyrate_limiter.AbstractBucket`
- Add your own `create_bucket` method in `tests/conftest.py` and pass it to the `create_bucket` fixture
- Run the test suite to validate the result
If the tests pass through, then you are just good to go with your new, fancy bucket!
Raw data
{
"_id": null,
"home_page": "https://github.com/vutran1710/PyrateLimiter",
"name": "pyrate-limiter",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "rate,rate-limiter,rate_limiter,ratelimiter,leaky-bucket,ratelimit,ratelimiting",
"author": "vutr",
"author_email": "me@vutr.io",
"download_url": "https://files.pythonhosted.org/packages/fa/06/13ffff42f7293f96ee45ec61cd50e2d27b39629652550383ffd212b802b1/pyrate_limiter-3.6.0.tar.gz",
"platform": null,
"description": "<img align=\"left\" width=\"95\" height=\"120\" src=\"https://raw.githubusercontent.com/vutran1710/PyrateLimiter/master/docs/_static/logo.png\">\n\n# PyrateLimiter\n\nThe request rate limiter using Leaky-bucket Algorithm.\n\nFull project documentation can be found at [pyratelimiter.readthedocs.io](https://pyratelimiter.readthedocs.io).\n\n[](https://badge.fury.io/py/pyrate-limiter)\n[](https://pypi.org/project/pyrate-limiter)\n[](https://codecov.io/gh/vutran1710/PyrateLimiter)\n[](https://github.com/vutran1710/PyrateLimiter/graphs/commit-activity)\n[](https://pypi.python.org/pypi/pyrate-limiter/)\n\n<br>\n\n## Contents\n\n- [PyrateLimiter](#pyratelimiter)\n - [Contents](#contents)\n - [Features](#features)\n - [Installation](#installation)\n - [Quickstart](#quickstart)\n - [Basic usage](#basic-usage)\n - [Key concepts](#key-concepts)\n - [Defining rate limits & buckets](#defining-rate-limits-and-buckets)\n - [Defining clock & routing logic](#defining-clock--routing-logic-with-bucketfactory)\n - [Wrapping all up with Limiter](#wrapping-all-up-with-limiter)\n - [Weight](#weight)\n - [Handling exceeded limits](#handling-exceeded-limits)\n - [Bucket analogy](#bucket-analogy)\n - [Rate limit exceptions](#rate-limit-exceptions)\n - [Rate limit delays](#rate-limit-delays)\n - [Backends](#backends)\n - [InMemoryBucket](#inmemorybucket)\n - [SQLiteBucket](#sqlitebucket)\n - [RedisBucket](#redisbucket)\n - [PostgresBucket](#postgresbucket)\n - [Decorator](#decorator)\n - [Advanced Usage](#advanced-usage)\n - [Component-level Diagram](#component-level-diagram)\n - [Time sources](#time-sources)\n - [Leaking](#leaking)\n - [Concurrency](#concurrency)\n - [Custom backend](#custom-backend)\n\n## Features\n\n- Tracks any number of rate limits and intervals you want to define\n- Independently tracks rate limits for multiple services or resources\n- Handles exceeded rate limits by either raising errors or adding delays\n- Several usage options including a normal function call, a decorator\n- Out-of-the-box workable with both sync & async\n- Includes optional SQLite and Redis backends, which can be used to persist limit tracking across\n multiple threads, or application restarts\n\n## Installation\n\n**PyrateLimiter** supports **python ^3.8**\n\nInstall using pip:\n\n```\npip install pyrate-limiter\n```\n\nOr using conda:\n\n```\nconda install --channel conda-forge pyrate-limiter\n```\n\n## Quickstart\n\nLet's say you want to limit 5 requests over 2 seconds, and raise an exception if the limit is exceeded:\n\n```python\nfrom pyrate_limiter import Duration, Rate, Limiter, BucketFullException\n\nrate = Rate(5, Duration.SECOND * 2)\nlimiter = Limiter(rate)\n\n# Or you can pass multiple rates\n# rates = [Rate(5, Duration.SECOND * 2), Rate(10, Duration.MINUTE)]\n# limiter = Limiter(rates)\n\nfor request in range(6):\n try:\n limiter.try_acquire(request)\n except BucketFullException as err:\n print(err)\n print(err.meta_info)\n# Bucket for item=5 with Rate limit=5/2.0s is already full\n# {'error': 'Bucket for item=5 with Rate limit=5/2.0s is already full', 'name': 5, 'weight': 1, 'rate': 'limit=5/2.0s'}\n```\n\n## Basic Usage\n\n### Key concepts\n\n#### Clock\n\n- Timestamp items\n\n#### Bucket\n\n- Hold items with timestamps.\n- Behave like a FIFO queue\n- It can `leak` - popping items that are no longer relevant out of the queue\n\n#### BucketFactory\n\n- BucketFactory keeps references to buckets & clocks: determine the exact time that items arrive then route them to their corresponding buckets\n- Help schedule background tasks to run buckets' `leak` periodically to make sure buckets will not explode from containing too many items\n- Where user define his own logic: routing, condition-checking, timing etc...\n\n#### Limiter\n\nThe Limiter's most important responsibility is to make user's life as easiest as possible:\n\n- Sums up all the underlying logic to a simple, intuitive API to work with\n- Handles async/sync context seamlessly (everything just `works` by adding/removing `async/await` keyword to the user's code)\n- Provides different ways of interacting with the underlying **BucketFactory** _(plain method call, decorator, context-manager (TBA))_\n- Provides thread-safety using RLock\n\n### Defining rate limits and buckets\n\nConsider some public API (like LinkedIn, GitHub, etc.) that has rate limits like the following:\n\n```\n- 500 requests per hour\n- 1000 requests per day\n- 10000 requests per month\n```\n\nYou can define these rates using the `Rate` class. `Rate` class has 2 properties only: **limit** and **interval**\n\n```python\nfrom pyrate_limiter import Duration, Rate\n\nhourly_rate = Rate(500, Duration.HOUR) # 500 requests per hour\ndaily_rate = Rate(1000, Duration.DAY) # 1000 requests per day\nmonthly_rate = Rate(10000, Duration.WEEK * 4) # 10000 requests per month\n\nrates = [hourly_rate, daily_rate, monthly_rate]\n```\n\nRates must be properly ordered:\n\n- Rates' intervals & limits must be ordered from least to greatest\n- Rates' ratio of **limit/interval** must be ordered from greatest to least\n\nExisting implementations of Bucket come with rate-validation when init. If you are to use your own implementation, use the validator provided by the lib\n\n```python\nfrom pyrate_limiter import validate_rate_list\n\nassert validate_rate_list(my_rates)\n```\n\nThen, add the rates to the bucket of your choices\n\n```python\nfrom pyrate_limiter import InMemoryBucket, RedisBucket\n\nbasic_bucket = InMemoryBucket(rates)\n\n# Or, using redis\nfrom redis import Redis\n\nredis_connection = Redis(host='localhost')\nredis_bucket = RedisBucket.init(rates, redis_connection, \"my-bucket-name\")\n\n# Async Redis would work too!\nfrom redis.asyncio import Redis\n\nredis_connection = Redis(host='localhost')\nredis_bucket = await RedisBucket.init(rates, redis_connection, \"my-bucket-name\")\n```\n\nIf you only need a single Bucket for everything, and python's built-in `time()` is enough for you, then pass the bucket to Limiter then ready to roll!\n\n```python\nfrom pyrate_limiter import Limiter\n\n# Limiter constructor accepts single bucket as the only parameter,\n# the rest are 3 optional parameters with default values as following\n# Limiter(bucket, clock=TimeClock(), raise_when_fail=True, max_delay=None)\nlimiter = Limiter(bucket)\n\n# Limiter is now ready to work!\nlimiter.try_acquire(\"hello world\")\n```\n\nIf you want to have finer grain control with routing & clocks etc, then you should use `BucketFactory`.\n\n### Defining Clock & routing logic with BucketFactory\n\nIf you need more than one type of Bucket, and be able to route items to different buckets based on some condition, you can use BucketFactory to do that.\n\nFrom the above steps, you already have your buckets. Now it's time to define what `Time` is (funny?!). Most of the time (again!?), you can use the existing Clock backend provided by **pyrate_limiter**.\n\n```python\nfrom pyrate_limiter.clock import TimeClock, MonotonicClock, SQLiteClock\n\nbase_clock = TimeClock()\n```\n\n**PyrateLimiter** makes no assumption about users logic, so to map coming items to their correct buckets, implement your own **BucketFactory** class! At minimum, there are only 2 methods require implementing\n\n```python\nfrom pyrate_limiter import BucketFactory\nfrom pyrate_limiter import AbstractBucket\n\n\nclass MyBucketFactory(BucketFactory):\n # You can use constructor here,\n # nor it requires to make bucket-factory work!\n\n def wrap_item(self, name: str, weight: int = 1) -> RateItem:\n \"\"\"Time-stamping item, return a RateItem\"\"\"\n now = clock.now()\n return RateItem(name, now, weight=weight)\n\n def get(self, _item: RateItem) -> AbstractBucket:\n \"\"\"For simplicity's sake, all items route to the same, single bucket\"\"\"\n return bucket\n```\n\n### Creating buckets dynamically\n\nIf more than one bucket is needed, the bucket-routing logic should go to BucketFactory `get(..)` method.\n\nWhen creating buckets dynamically, it is needed to schedule leak for each newly created buckets.\n\nTo support this, BucketFactory comes with a predefined method call `self.create(..)`. It is meant to create the bucket and schedule that bucket for leaking using the Factory's clock\n\n```python\ndef create(\n self,\n clock: AbstractClock,\n bucket_class: Type[AbstractBucket],\n *args,\n **kwargs,\n ) -> AbstractBucket:\n \"\"\"Creating a bucket dynamically\"\"\"\n bucket = bucket_class(*args, **kwargs)\n self.schedule_leak(bucket, clock)\n return bucket\n```\n\nBy utilizing this, we can modify the code as following:\n\n```python\nclass MultiBucketFactory(BucketFactory):\n def __init__(self, clock):\n self.clock = clock\n self.buckets = {}\n\n def wrap_item(self, name: str, weight: int = 1) -> RateItem:\n \"\"\"Time-stamping item, return a RateItem\"\"\"\n now = clock.now()\n return RateItem(name, now, weight=weight)\n\n def get(self, item: RateItem) -> AbstractBucket:\n if item.name not in self.buckets:\n # Use `self.create(..)` method to both initialize new bucket and calling `schedule_leak` on that bucket\n # We can create different buckets with different types/classes here as well\n new_bucket = self.create(YourBucketClass, *your-arguments, **your-keyword-arguments)\n self.buckets.update({item.name: new_bucket})\n\n return self.buckets[item.name]\n```\n\n### Wrapping all up with Limiter\n\nPass your bucket-factory to Limiter, and ready to roll!\n\n```python\nfrom pyrate_limiter import Limiter\n\nlimiter = Limiter(\n bucket_factory,\n raise_when_fail=False, # Default = True\n max_delay=1000, # Default = None\n)\n\nitem = \"the-earth\"\nlimiter.try_acquire(item)\n\nheavy_item = \"the-sun\"\nlimiter.try_acquire(heavy_item, weight=10000)\n```\n\nIf your bucket's backend is `async`, well, we got you covered! Passing `await` to the limiter is enought to make it scream!\n\n```python\nawait limiter.try_acquire(item)\n```\n\nAlternatively, you can use `Limiter.try_acquire` as a function decorator. But you have to provide a `mapping` function that map the wrapped function's arguments to a proper `limiter.try_acquire` argument - which is a tuple of `(str, int)` or just `str`\n\n```python\nmy_beautiful_decorator = limiter.as_decorator()\n\ndef mapping(some_number: int):\n return str(some_number)\n\n@my_beautiful_decorator(mapping)\ndef request_function(some_number: int):\n requests.get('https://example.com')\n\n# Async would work too!\n@my_beautiful_decorator(mapping)\nasync def async_request_function(some_number: int):\n requests.get('https://example.com')\n```\n\n### Weight\n\nItem can have weight. By default item's weight = 1, but you can modify the weight before passing to `limiter.try_acquire`.\n\nItem with weight W > 1 when consumed will be multiplied to (W) items with the same timestamp and weight = 1. Example with a big item with weight W=5, when put to bucket, it will be divided to 5 items with weight=1 + following names\n\n```\nBigItem(weight=5, name=\"item\", timestamp=100) => [\n item(weight=1, name=\"item\", timestamp=100),\n item(weight=1, name=\"item\", timestamp=100),\n item(weight=1, name=\"item\", timestamp=100),\n item(weight=1, name=\"item\", timestamp=100),\n item(weight=1, name=\"item\", timestamp=100),\n]\n```\n\nYet, putting this big, heavy item into bucket is expected to be transactional & atomic - meaning either all 5 items will be consumed or none of them will. This is made possible as bucket `put(item)` always check for available space before ingesting. All of the Bucket's implementations provided by **PyrateLimiter** follows this rule.\n\nAny additional, custom implementation of Bucket are expected to behave alike - as we have unit tests to cover the case.\n\nSee [Advanced usage options](#advanced-usage) below for more details.\n\n### Handling exceeded limits\n\nWhen a rate limit is exceeded, you have two options: raise an exception, or add delays.\n\n#### Bucket analogy\n\n<img height=\"300\" align=\"right\" src=\"https://upload.wikimedia.org/wikipedia/commons/c/c4/Leaky_bucket_analogy.JPG\">\n\nAt this point it's useful to introduce the analogy of \"buckets\" used for rate-limiting. Here is a\nquick summary:\n\n- This library implements the [Leaky Bucket algorithm](https://en.wikipedia.org/wiki/Leaky_bucket).\n- It is named after the idea of representing some kind of fixed capacity -- like a network or service -- as a bucket.\n- The bucket \"leaks\" at a constant rate. For web services, this represents the **ideal or permitted request rate**.\n- The bucket is \"filled\" at an intermittent, unpredicatble rate, representing the **actual rate of requests**.\n- When the bucket is \"full\", it will overflow, representing **canceled or delayed requests**.\n- Item can have weight. Consuming a single item with weight W > 1 is the same as consuming W smaller, unit items - each with weight=1, with the same timestamp and maybe same name (depending on however user choose to implement it)\n\n#### Rate limit exceptions\n\nBy default, a `BucketFullException` will be raised when a rate limit is exceeded.\nThe error contains a `meta_info` attribute with the following information:\n\n- `name`: The name of item it received\n- `weight`: The weight of item it received\n- `rate`: The specific rate that has been exceeded\n\nHere's an example that will raise an exception on the 4th request:\n\n```python\nrate = Rate(3, Duration.SECOND)\nbucket = InMemoryBucket([rate])\nclock = TimeClock()\n\n\nclass MyBucketFactory(BucketFactory):\n\n def wrap_item(self, name: str, weight: int = 1) -> RateItem:\n \"\"\"Time-stamping item, return a RateItem\"\"\"\n now = clock.now()\n return RateItem(name, now, weight=weight)\n\n def get(self, _item: RateItem) -> AbstractBucket:\n \"\"\"For simplicity's sake, all items route to the same, single bucket\"\"\"\n return bucket\n\n\nlimiter = Limiter(MyBucketFactory())\n\nfor _ in range(4):\n try:\n limiter.try_acquire('item', weight=2)\n except BucketFullException as err:\n print(err)\n # Output: Bucket with Rate 3/1.0s is already full\n print(err.meta_info)\n # Output: {'name': 'item', 'weight': 2, 'rate': '3/1.0s', 'error': 'Bucket with Rate 3/1.0s is already full'}\n```\n\nThe rate part of the output is constructed as: `limit / interval`. On the above example, the limit\nis 3 and the interval is 1, hence the `Rate 3/1`.\n\n#### Rate limit delays\n\nYou may want to simply slow down your requests to stay within the rate limits instead of canceling\nthem. In that case you pass the `max_delay` argument the maximum value of delay (typically in _ms_ when use human-clock).\n\n```python\nlimiter = Limiter(factory, max_delay=500) # Allow to delay up to 500ms\n```\n\nAs `max_delay` has been passed as a numeric value, when ingesting item, limiter will:\n\n- First, try to ingest such item using the routed bucket\n- If it fails to put item into the bucket, it will call `wait(item)` on the bucket to see how much time remains until the bucket can consume the item again?\n- Comparing the `wait` value to the `max_delay`.\n- if `max_delay` >= `wait`: delay (wait + 50ms as latency-tolerance) using either `asyncio.sleep` or `time.sleep` until the bucket can consume again\n- if `max_delay` < `wait`: it raises `LimiterDelayException` if Limiter's `raise_when_fail=True`, otherwise silently fail and return False\n\nExample:\n\n```python\nfrom pyrate_limiter import LimiterDelayException\n\nfor _ in range(4):\n try:\n limiter.try_acquire('item', weight=2, max_delay=200)\n except LimiterDelayException as err:\n print(err)\n # Output:\n # Actual delay exceeded allowance: actual=500, allowed=200\n # Bucket for 'item' with Rate 3/1.0s is already full\n print(err.meta_info)\n # Output: {'name': 'item', 'weight': 2, 'rate': '3/1.0s', 'max_delay': 200, 'actual_delay': 500}\n```\n\n### Backends\n\nA few different bucket backends are available:\n\n- **InMemoryBucket**: using python built-in list as bucket\n- **RedisBucket**, using err... redis, with both async/sync support\n- **PostgresBucket**, using `psycopg2`\n- **SQLiteBucket**, using sqlite3\n\n#### InMemoryBucket\n\nThe default bucket is stored in memory, using python `list`\n\n```python\nfrom pyrate_limiter import InMemoryBucket, Rate, Duration\n\nrates = [Rate(5, Duration.MINUTE * 2)]\nbucket = InMemoryBucket(rates)\n```\n\nThis bucket only availabe in `sync` mode. The only constructor argument is `List[Rate]`.\n\n#### RedisBucket\n\nRedisBucket uses `Sorted-Set` to store items with key being item's name and score item's timestamp\nBecause it is intended to work with both async & sync, we provide a classmethod `init` for it\n\n```python\nfrom pyrate_limiter import RedisBucket, Rate, Duration\n\n# Using synchronous redis\nfrom redis import ConnectionPool\nfrom redis import Redis\n\nrates = [Rate(5, Duration.MINUTE * 2)]\npool = ConnectionPool.from_url(\"redis://localhost:6379\")\nredis_db = Redis(connection_pool=pool)\nbucket_key = \"bucket-key\"\nbucket = RedisBucket.init(rates, redis_db, bucket_key)\n\n# Using asynchronous redis\nfrom redis.asyncio import ConnectionPool as AsyncConnectionPool\nfrom redis.asyncio import Redis as AsyncRedis\n\npool = AsyncConnectionPool.from_url(\"redis://localhost:6379\")\nredis_db = AsyncRedis(connection_pool=pool)\nbucket_key = \"bucket-key\"\nbucket = await RedisBucket.init(rates, redis_db, bucket_key)\n```\n\nThe API are the same, regardless of sync/async. If AsyncRedis is being used, calling `await bucket.method_name(args)` would just work!\n\n#### SQLiteBucket\n\nIf you need to persist the bucket state, a SQLite backend is available.\n\nManully create a connection to Sqlite and pass it along with the table name to the bucket class:\n\n```python\nfrom pyrate_limiter import SQLiteBucket, Rate, Duration\nimport sqlite3\n\nrates = [Rate(5, Duration.MINUTE * 2)]\nconn = sqlite3.connect(\n \"/var/mydb.sqlite\",\n isolation_level=\"EXCLUSIVE\",\n check_same_thread=False,\n)\ntable = \"my-bucket-table\"\nbucket = SQLiteBucket(rates, conn, table)\n```\n\n#### PostgresBucket\n\nPostgres is supported, but you have to install `psycopg[pool]` either as an extra or as a separate package.\n\nYou can use Postgres's built-in **CURRENT_TIMESTAMP** as the time source with `PostgresClock`, or use an external custom time source.\n\n```python\nfrom pyrate_limiter import PostgresBucket, Rate, PostgresClock\nfrom psycopg_pool import ConnectionPool\n\nconnection_pool = ConnectionPool('postgresql://postgres:postgres@localhost:5432')\n\nclock = PostgresClock(connection_pool)\nrates = [Rate(3, 1000), Rate(4, 1500)]\nbucket = PostgresBucket(connection_pool, \"my-bucket-table\", rates)\n```\n\n### Decorator\n\nLimiter can be used as decorator, but you have to provide a `mapping` function that maps the wrapped function's arguments to `limiter.try_acquire` function arguments. The mapping function must return either a tuple of `(str, int)` or just a `str`\n\nThe decorator can work with both sync & async function\n\n```python\ndecorator = limiter.as_decorator()\n\ndef mapping(*args, **kwargs):\n return \"demo\", 1\n\n@decorator(mapping)\ndef handle_something(*args, **kwargs):\n \"\"\"function logic\"\"\"\n\n@decorator(mapping)\nasync def handle_something_async(*args, **kwargs):\n \"\"\"function logic\"\"\"\n```\n\n## Advanced Usage\n\n### Component level diagram\n\n\n\n### Time sources\n\nTime source can be anything from anywhere: be it python's built-in time, or monotonic clock, sqliteclock, or crawling from world time server(well we don't have that, but you can!).\n\n```python\nfrom pyrate_limiter import TimeClock # use python' time.time()\nfrom pyrate_limiter import MonotonicClock # use python time.monotonic()\n```\n\nClock's abstract interface only requires implementing a method `now() -> int`. And it can be both sync or async.\n\n### Leaking\n\nTypically bucket should not hold items forever. Bucket's abstract interface requires its implementation must be provided with `leak(current_timestamp: Optional[int] = None)`.\n\nThe `leak` method when called is expected to remove any items considered outdated at that moment. During Limiter lifetime, all the buckets' `leak` should be called periodically.\n\n**BucketFactory** provide a method called `schedule_leak` to help deal with this matter. Basically, it will run as a background task for all the buckets currently in use, with interval between `leak` call by **default is 10 seconds**.\n\n```python\n# Runnning a background task (whether it is sync/async - doesnt matter)\n# calling the bucket's leak\nfactory.schedule_leak(bucket, clock)\n```\n\nYou can change this calling interval by overriding BucketFactory's `leak_interval` property. This interval is in **miliseconds**.\n\n```python\nclass MyBucketFactory(BucketFactory):\n def __init__(self, *args):\n self.leak_interval = 300\n```\n\nWhen dealing with leak using BucketFactory, the author's suggestion is, we can be pythonic about this by implementing a constructor\n\n```python\nclass MyBucketFactory(BucketFactory):\n\n def constructor(self, clock, buckets):\n self.clock = clock\n self.buckets = buckets\n\n for bucket in buckets:\n self.schedule_leak(bucket, clock)\n\n```\n\n### Concurrency\n\nGenerally, Lock is provided at Limiter's level, except SQLiteBucket case.\n\n### Custom backends\n\nIf these don't suit your needs, you can also create your own bucket backend by implementing `pyrate_limiter.AbstractBucket` class.\n\nOne of **PyrateLimiter** design goals is powerful extensibility and maximum ease of development.\n\nIt must be not only be a ready-to-use tool, but also a guide-line, or a framework that help implementing new features/bucket free of the most hassles.\n\nDue to the composition nature of the library, it is possbile to write minimum code and validate the result:\n\n- Fork the repo\n- Implement your bucket with `pyrate_limiter.AbstractBucket`\n- Add your own `create_bucket` method in `tests/conftest.py` and pass it to the `create_bucket` fixture\n- Run the test suite to validate the result\n\nIf the tests pass through, then you are just good to go with your new, fancy bucket!\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python Rate-Limiter using Leaky-Bucket Algorithm",
"version": "3.6.0",
"project_urls": {
"Documentation": "https://pyrate-limiter.readthedocs.io",
"Homepage": "https://github.com/vutran1710/PyrateLimiter",
"Repository": "https://github.com/vutran1710/PyrateLimiter"
},
"split_keywords": [
"rate",
"rate-limiter",
"rate_limiter",
"ratelimiter",
"leaky-bucket",
"ratelimit",
"ratelimiting"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a27678124018c5306331961cc43d2ba301d9d3d56610c938b24c828b94751ff7",
"md5": "d8a3a6973c441587e61ece9ab12e6ac5",
"sha256": "715e9f08c6fe2a00d0cae5b1a6647d68ffeb2a54dc0cc2cff4e046b067ce6da4"
},
"downloads": -1,
"filename": "pyrate_limiter-3.6.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d8a3a6973c441587e61ece9ab12e6ac5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 26686,
"upload_time": "2024-03-17T15:06:46",
"upload_time_iso_8601": "2024-03-17T15:06:46.704491Z",
"url": "https://files.pythonhosted.org/packages/a2/76/78124018c5306331961cc43d2ba301d9d3d56610c938b24c828b94751ff7/pyrate_limiter-3.6.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "fa0613ffff42f7293f96ee45ec61cd50e2d27b39629652550383ffd212b802b1",
"md5": "42c97604de98b1c4eb419f35924adf65",
"sha256": "390f97066b322732e498c9e921fbdfd31d9ec0070a14e06da9af0efc62e091e4"
},
"downloads": -1,
"filename": "pyrate_limiter-3.6.0.tar.gz",
"has_sig": false,
"md5_digest": "42c97604de98b1c4eb419f35924adf65",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 280021,
"upload_time": "2024-03-17T15:06:48",
"upload_time_iso_8601": "2024-03-17T15:06:48.904354Z",
"url": "https://files.pythonhosted.org/packages/fa/06/13ffff42f7293f96ee45ec61cd50e2d27b39629652550383ffd212b802b1/pyrate_limiter-3.6.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-17 15:06:48",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "vutran1710",
"github_project": "PyrateLimiter",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pyrate-limiter"
}
JSON
