token-throttler 1.2.0

Token throttler is an extendable rate-limiting library somewhat based on a token bucket algorithm

Token throttler

Token throttler is an extendable rate-limiting library somewhat based on a token bucket algorithm.

Table of contents

1. Installation
2. Features
3. Usage
   1. Manual usage example
   2. Decorator usage example
4. Storage
5. Configuration
   1. Configuration usage

1. Installation

Token throttler is available on PyPI:

$ python -m pip install token-throttler

Token throttler officially supports Python >= 3.7.

NOTE: Depending on the storage engine you pick, you can install token throttler with extras:

$ python -m pip install token-throttler[redis]
$ python -m pip install token-throttler[aioredis]

2. Features

• Blocking (TokenThrottler) and non-blocking (TokenThrottlerAsync)
• Global throttler(s) configuration
• Configurable token throttler cost and identifier
• Multiple buckets per throttler per identifier
• Buckets can be added/removed manually or by a dict configuration
• Manual usage or usage via decorator
• Decorator usage supports async code too
• Custom decorator can be written
• Extendable storage engine (eg. Redis)

3. Usage

Token throttler supports both manual usage and via decorator.

Decorator usage supports both async and sync.

1) Manual usage example:

from token_throttler import TokenBucket, TokenThrottler
from token_throttler.storage import RuntimeStorage

throttler: TokenThrottler = TokenThrottler(cost=1, storage=RuntimeStorage())
throttler.add_bucket(identifier="hello_world", bucket=TokenBucket(replenish_time=10, max_tokens=10))
throttler.add_bucket(identifier="hello_world", bucket=TokenBucket(replenish_time=30, max_tokens=20))


def hello_world() -> None:
    print("Hello World")


for i in range(10):
    throttler.consume(identifier="hello_world")
    hello_world()

if throttler.consume(identifier="hello_world"):
    hello_world()
else:
    print("bucket_one ran out of tokens")

2) Decorator usage example:

from token_throttler import TokenBucket, TokenThrottler, TokenThrottlerException
from token_throttler.storage import RuntimeStorage

throttler: TokenThrottler = TokenThrottler(1, RuntimeStorage())
throttler.add_bucket("hello_world", TokenBucket(10, 10))
throttler.add_bucket("hello_world", TokenBucket(30, 20))


@throttler.enable("hello_world")
def hello_world() -> None:
    print("Hello World")


for i in range(10):
    hello_world()

try:
    hello_world()
except TokenThrottlerException:
    print("bucket_one ran out of tokens")

For other examples see examples directory.

4. Storage

TokenThrottler supports RuntimeStorage and RedisStorage. TokenThrottlerAsync supports RedisStorageAsync

If you want your own storage engine, feel free to extend the token_throttler.storage.BucketStorage or token_throttler.storage.BucketStorageAsync classes.

For storage examples see examples directory.

5. Configuration

Token throttler supports global configuration by making use of ThrottlerConfig class.

Configuration params:

• IDENTIFIER_FAIL_SAFE - if invalid identifier is given as a param for the consume method and IDENTIFIER_FAIL_SAFE is set to True, no KeyError exception will be raised and consume will act like a limitless bucket is being consumed.
• ENABLE_THREAD_LOCK - if set to True, throttler will acquire a thread lock upon calling consume method and release the lock once the consume is finished. This avoids various race conditions at a slight performance cost.

Configuration usage

from token_throttler import ThrottlerConfig, TokenBucket, TokenThrottler
from token_throttler.storage import RuntimeStorage

ThrottlerConfig.set({
   "ENABLE_THREAD_LOCK": False,
   "IDENTIFIER_FAIL_SAFE": True,
})
throttler: TokenThrottler = TokenThrottler(1, RuntimeStorage())
throttler.add_bucket("hello_world", TokenBucket(10, 10))
throttler.add_bucket("hello_world", TokenBucket(30, 20))
...