pyrate-limiter-concord 2.9.3

Description:

pyratelimiterconcord 2.9.3

PyrateLimiter
The request rate limiter using Leaky-bucket algorithm.
Full project documentation can be found at pyratelimiter.readthedocs.io.






Contents

PyrateLimiter

Contents
Features
Installation
Basic usage

Defining rate limits
Applying rate limits
Identities


Handling exceeded limits

Bucket analogy
Rate limit exceptions
Rate limit delays


Additional usage options

Decorator
Contextmanager
Async decorator/contextmanager


Backends

Memory
SQLite
Redis
Custom backends


Additional features

Time sources


Examples



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, or a contextmanager
Async support
Includes optional SQLite and Redis backends, which can be used to persist limit tracking across
multiple threads, processes, or application restarts

Installation
Install using pip:
pip install pyrate-limiter

Or using conda:
conda install --channel conda-forge pyrate-limiter

Basic usage
Defining rate limits
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 RequestRate class, and add them to a Limiter:
from pyrate_limiter import BucketFullException, Duration, RequestRate, Limiter

hourly_rate = RequestRate(500, Duration.HOUR) # 500 requests per hour
daily_rate = RequestRate(1000, Duration.DAY) # 1000 requests per day
monthly_rate = RequestRate(10000, Duration.MONTH) # 10000 requests per month

limiter = Limiter(hourly_rate, daily_rate, monthly_rate)

Note that these rates need to be ordered by interval length; in other words, an hourly rate must
come before a daily rate, etc.
Applying rate limits
Then, use Limiter.try_acquire() wherever you are making requests (or other rate-limited operations).
This will raise an exception if the rate limit is exceeded.
import requests

def request_function():
limiter.try_acquire('identity')
requests.get('https://example.com')

while True:
request_function()

Alternatively, you can use Limiter.ratelimit() as a function decorator:
@limiter.ratelimit('identity')
def request_function():
requests.get('https://example.com')

See Additional usage options below for more details.
Identities
Note that both try_acquire() and ratelimit() take one or more identity arguments. Typically this is
the name of the service or resource that is being rate-limited. This allows you to track rate limits
for these resources independently. For example, if you have a service that is rate-limited by user:
def request_function(user_ids):
limiter.try_acquire(*user_ids)
for user_id in user_ids:
requests.get(f'https://example.com?user_id={user_id}')

Handling exceeded limits
When a rate limit is exceeded, you have two options: raise an exception, or add delays.
Bucket analogy

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

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:

identity: The identity it received
rate: The specific rate that has been exceeded
remaining_time: The remaining time until the next request can be sent

Here's an example that will raise an exception on the 4th request:
limiter = Limiter(RequestRate(3, Duration.SECOND))

for _ in range(4):
try:
limiter.try_acquire('vutran')
except BucketFullException as err:
print(err)
# Output: Bucket for vutran with Rate 3/5 is already full
print(err.meta_info)
# Output: {'identity': 'vutran', 'rate': '5/5', 'remaining_time': 2.9,
# 'error': 'Bucket for vutran with Rate 3/5 is already full'}

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 can use the delay argument. Note that this is only available for
Limiter.ratelimit():
@limiter.ratelimit('identity', delay=True)
def my_function():
do_stuff()

If you exceed a rate limit with a long interval (daily, monthly, etc.), you may not want to delay
that long. In this case, you can set a max_delay (in seconds) that you are willing to wait in
between calls:
@limiter.ratelimit('identity', delay=True, max_delay=360)
def my_function():
do_stuff()

In this case, calls may be delayed by at most 360 seconds to stay within the rate limits; any longer
than that, and a BucketFullException will be raised instead. Without specifying max_delay, calls
will be delayed as long as necessary.
Additional usage options
Besides Limiter.try_acquire(), some additional usage options are available using Limiter.ratelimit():
Decorator
Limiter.ratelimit() can be used as a decorator:
@limiter.ratelimit('identity')
def my_function():
do_stuff()

As with Limiter.try_acquire(), if calls to the wrapped function exceed the rate limits you
defined, a BucketFullException will be raised.
Contextmanager
Limiter.ratelimit() also works as a contextmanager:
def my_function():
with limiter.ratelimit('identity', delay=True):
do_stuff()

Async decorator/contextmanager
Limiter.ratelimit() also support async functions, either as a decorator or contextmanager:
@limiter.ratelimit('identity', delay=True)
async def my_function():
await do_stuff()

async def my_function():
async with limiter.ratelimit('identity'):
await do_stuff()

When delays are enabled for an async function, asyncio.sleep() will be used instead of time.sleep().
Backends
A few different bucket backends are available, which can be selected using the bucket_class
argument for Limiter. Any additional backend-specific arguments can be passed
via bucket_kwargs.
Memory
The default bucket is stored in memory, backed by a queue.Queue. A list implementation is also available:
from pyrate_limiter import Limiter, MemoryListBucket

limiter = Limiter(bucket_class=MemoryListBucket)

SQLite
If you need to persist the bucket state, a SQLite backend is available.
By default it will store the state in the system temp directory, and you can use
the path argument to use a different location:
from pyrate_limiter import Limiter, SQLiteBucket

limiter = Limiter(bucket_class=SQLiteBucket)

By default, the database will be stored in the system temp directory. You can specify a different
path via bucket_kwargs:
limiter = Limiter(
bucket_class=SQLiteBucket,
bucket_kwargs={'path': '/path/to/db.sqlite'},
)

Concurrency
This backend is thread-safe, and may also be used with multiple child processes that share the same
Limiter object, e.g. if created with ProcessPoolExecutor or multiprocessing.Process.
If you want to use SQLite with multiple processes with no shared state, for example if created by
running multiple scripts or by an external process, some additional protections are needed. For
these cases, a separate FileLockSQLiteBucket class is available. This requires installing the
py-filelock library.
limiter = Limiter(bucket_class=FileLockSQLiteBucket)

Redis
If you have a larger, distributed application, Redis is an ideal backend. This
option requires redis-py.
Note that this backend requires a bucket_name argument, which will be used as a kprefix for the
Redis keys created. This can be used to disambiguate between multiple services using the same Redis
instance with pyrate-limiter.
from pyrate_limiter import Limiter, RedisBucket

limiter = Limiter(bucket_class=RedisBucket, bucket_kwargs={'bucket_name': 'my_service'})

Connection settings
If you need to pass additional connection settings, you can use the redis_pool bucket argument:
from redis import ConnectionPool

redis_pool = ConnectionPool(host='localhost', port=6379, db=0)
limiter = Limiter(
bucket_class=RedisBucket,
bucket_kwargs={'redis_pool': redis_pool, 'bucket_name': 'my_service'},
)

Redis clusters
Redis clusters are also supported, which requires
redis-py-cluster:
from pyrate_limiter import Limiter, RedisClusterBucket

limiter = Limiter(bucket_class=RedisClusterBucket)

Custom backends
If these don't suit your needs, you can also create your own bucket backend by extending pyrate_limiter.bucket.AbstractBucket.
Additional features
Time sources
By default, monotonic time is used, to ensure requests are always logged in the correct order.
You can specify a custom time source with the time_function argument. For example, you may want to
use the current UTC time for consistency across a distributed application using a Redis backend.
from datetime import datetime
from pyrate_limiter import Duration, Limiter, RequestRate
from time import time

rate = RequestRate(5, Duration.SECOND)
limiter_datetime = Limiter(rate, time_function=lambda: datetime.utcnow().timestamp())

Or simply use the basic time.time() function:
from time import time

rate = RequestRate(5, Duration.SECOND)
limiter_time = Limiter(rate, time_function=time)

Examples
To prove that pyrate-limiter is working as expected, here is a complete example to demonstrate
rate-limiting with delays:
from time import perf_counter as time
from pyrate_limiter import Duration, Limiter, RequestRate

limiter = Limiter(RequestRate(5, Duration.SECOND))
n_requests = 27

@limiter.ratelimit("test", delay=True)
def limited_function(start_time):
print(f"t + {(time() - start_time):.5f}")

start_time = time()
for _ in range(n_requests):
limited_function(start_time)
print(f"Ran {n_requests} requests in {time() - start_time:.5f} seconds")

And an equivalent example for async usage:
import asyncio
from time import perf_counter as time
from pyrate_limiter import Duration, Limiter, RequestRate

limiter = Limiter(RequestRate(5, Duration.SECOND))
n_requests = 27

@limiter.ratelimit("test", delay=True)
async def limited_function(start_time):
print(f"t + {(time() - start_time):.5f}")

async def test_ratelimit():
start_time = time()
tasks = [limited_function(start_time) for _ in range(n_requests)]
await asyncio.gather(*tasks)
print(f"Ran {n_requests} requests in {time() - start_time:.5f} seconds")

asyncio.run(test_ratelimit())