API Reference

Simple Token Bucket

class simple_token_bucket.SimpleTokenBucket

__init__(name: str, bucket_size: int, refresh_interval: int, backend: Backend, listener: Callable[[int], None] | None = None)

Create a new SimpleTokenBucket.

Every refresh_interval the token bucket is refresh and bucket_size tokens are available again.

Parameters:

• name – Bucket name.

• bucket_size – Number of available tokens.

• refresh_interval – Bucket reset time interval (in seconds).

• backend – a backends.Backend implementation.

try_get_token(raises=True) → bool

exception simple_token_bucket.NotEnoughTokens

__init__(*, remaining_seconds: int)

class simple_token_bucket.backends.Backend

get_token(*, bucket_name: str, bucket_size: int, refresh_interval: int) → Tuple[int, int]

Returns available tokens in bucket_name and its time-to-live ( number of seconds before the next refresh).

Parameters:

• bucket_name – Bucket name.

• bucket_size – Default size of the bucket.

• refresh_interval – Bucket reset time interval (in seconds).

Returns:

• int: number of tokens available.

• int: nubmer of seconds before the next refresh.

Return type:

2-element tuple containing

class simple_token_bucket.backends.redis.RedisBackend

classmethod from_url(redis_url: str)

__init__(redis_client: Redis)

get_token(*, bucket_name: str, bucket_size: int, refresh_interval: int) → Tuple[int, int]

Returns available tokens in bucket_name and its time-to-live ( number of seconds before the next refresh).

Parameters:

• bucket_name – Bucket name.

• bucket_size – Default size of the bucket.

• refresh_interval – Bucket reset time interval (in seconds).

Returns:

• int: number of tokens available.

• int: nubmer of seconds before the next refresh.

Return type:

2-element tuple containing