`redditwarp.util.token_bucket`#

An implementation of the token bucket algorithm.

class redditwarp.util.token_bucket.TokenBucket(capacity: float, rate: float, *, time_func: ~typing.Callable[[], float] = <built-in function monotonic>)[source]#

Bases: object

capacity: float#

rate: float#

time_func: Callable[[], float]#

get_value() → float[source]#: Return the number of tokens in the bucket.

can_consume(n: float) → bool[source]#: Check if n tokens are available.

try_consume(n: float) → bool[source]#: Consume n tokens if n tokens are available.

consume(n: float) → None[source]#: Consume n tokens. If n tokens aren’t available, raise an exception.

hard_consume(n: float) → bool[source]#: Consume up to n tokens.

consume_all() → None[source]#: Empty the token bucket completely.

get_cooldown(n: float) → float[source]#: Return the duration the client should wait before the consume methods return True again.