Token Throttler

1.5.1 · active · verified Thu Apr 16

Token throttler is an extendable rate-limiting library for Python, somewhat based on the token bucket algorithm. It supports both blocking (sync) and non-blocking (async) operations, global and instance-specific configurations, and various storage backends including in-memory and Redis. The current version is 1.5.1, actively maintained with regular updates.

Common errors

```
KeyError: 'Bucket for identifier 'my_id' not found'
```
cause Attempted to consume tokens for an identifier that has no associated token bucket, and `IDENTIFIER_FAIL_SAFE` is set to `False` (the default).

fix Before calling `throttler.consume('my_id')`, ensure you've called `throttler.add_bucket('my_id', TokenBucket(max_tokens=X, replenish_time=Y))` or set `IDENTIFIER_FAIL_SAFE=True` in the throttler's `ThrottlerConfig`.
```
RuntimeWarning: Modifications on default_config object during runtime are not recommended
```
cause The global `default_config` singleton was modified after `TokenThrottler` instances that rely on it were already created. This can lead to unpredictable throttling behavior.

fix Perform all modifications to `default_config` at the very start of your application, before any `TokenThrottler` instances are initialized. For dynamic configuration, use instance-specific `ThrottlerConfig` objects when creating `TokenThrottler` instances.
```
My application is unexpectedly hitting rate limits (429 errors) or not throttling correctly.
```
cause The `TokenBucket` parameters (`max_tokens`, `replenish_time`) or the `cost` specified in `throttler.consume()` are not correctly configured for the desired rate limit, or not considering concurrent access without `ENABLE_THREAD_LOCK`.

fix Review your `TokenBucket` configurations. `max_tokens` determines burst capacity, `replenish_time` determines the rate at which tokens are refilled. Also, check the `cost` parameter in `consume()` calls. For multi-threaded scenarios, consider enabling `ENABLE_THREAD_LOCK=True` in your `ThrottlerConfig` to prevent token overconsumption due to race conditions.

Warnings

gotcha Modifying the `default_config` (global configuration) object at runtime is not recommended and emits a `RuntimeWarning`.
Fix: Configure global settings using `default_config.set()` *before* initializing any `TokenThrottler` instances. For instance-specific overrides, pass a `ThrottlerConfig` object to the `TokenThrottler` constructor instead.
gotcha In multi-threaded applications, race conditions can occur. The library offers `ENABLE_THREAD_LOCK` to mitigate this.
Fix: Set `ENABLE_THREAD_LOCK=True` in your `ThrottlerConfig` (globally or per instance) to enable a thread lock for the `consume` method, preventing race conditions. Be aware this incurs a slight performance cost.
gotcha Calling `throttler.consume()` with an unknown `identifier` will raise a `KeyError` by default.
Fix: Ensure all identifiers used in `consume` have a corresponding `TokenBucket` added via `throttler.add_bucket()`. Alternatively, set `IDENTIFIER_FAIL_SAFE=True` in `ThrottlerConfig` to make unknown identifiers act as limitless buckets, preventing `KeyError`.

Install

pip install token-throttler Basic Installation
pip install token-throttler[redis] With Redis Storage

Imports

TokenThrottler

from token_throttler import TokenThrottler

TokenThrottlerAsync

from token_throttler import TokenThrottlerAsync

TokenBucket
```
from token_throttler import TokenBucket
```

ThrottlerConfig

from token_throttler import ThrottlerConfig

default_config
```
from token_throttler import default_config
```
Singleton global configuration object.
RuntimeStorage
```
from token_throttler.storage import RuntimeStorage
```
In-memory storage for TokenThrottler.

Quickstart

This quickstart demonstrates initializing `TokenThrottler` with `RuntimeStorage` and adding a basic `TokenBucket`. It then simulates several API calls, showing how to consume tokens and handle rate limits. It also includes an example of consuming tokens with a custom cost.

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

# Initialize a throttler with a default cost of 1 token per consumption
# and in-memory storage.
throttler: TokenThrottler = TokenThrottler(cost=1, storage=RuntimeStorage())

# Add a token bucket named 'api_limit'
# It can hold 10 tokens and replenishes 10 tokens every 60 seconds (1 token per 6s).
throttler.add_bucket(
    identifier="api_limit",
    bucket=TokenBucket(replenish_time=60, max_tokens=10)
)

def make_api_call():
    if throttler.consume("api_limit"):
        print("API call allowed. Remaining tokens for 'api_limit'.")
        # Simulate actual API call
    else:
        print("API call denied. Rate limit exceeded for 'api_limit'.")

for i in range(15):
    print(f"Attempt {i+1}: ", end="")
    make_api_call()

# Example of using a specific cost for consumption
throttler.add_bucket("heavy_operation", TokenBucket(replenish_time=30, max_tokens=5))
if throttler.consume("heavy_operation", cost=2):
    print("Heavy operation allowed (cost 2 tokens).")
else:
    print("Heavy operation denied (cost 2 tokens).")

view raw JSON →