pyrate_limiter.buckets package

Concrete bucket implementations

class pyrate_limiter.buckets.InMemoryBucket(rates, algorithm=None)

Bases: AbstractBucket

Simple In-memory Bucket using native list Clock can be either time.time or time.monotonic When leak, clock is required Pros: fast, safe, and precise Cons: since it resides in local memory, the data is not persistent, nor scalable Usecase: small applications, simple logic

count()

Count number of items in the bucket

Return type:

int

flush()

Flush the whole bucket - Must remove failing-rate after flushing

Return type:

None

is_async = False
items
leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int

peek(index)

Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

Return type:

RateItem | None

put(item)

Put an item (typically the current time) in the bucket return true if successful, otherwise false

Return type:

bool

class pyrate_limiter.buckets.InMemoryStateStore

Bases: StateStore

State in a local attribute, guarded by a reentrant lock.

check(algorithm, rates, now, weight)

Apply algorithm.step to the stored state, atomically.

is_async = False

None means “ask the Leaker to probe” (a client that may be either).

read(algorithm, rates)

Current state. For reporting only - never the basis of a decision.

reset()

Forget everything, as though the key had never been used.

Return type:

None

class pyrate_limiter.buckets.MultiprocessBucket(rates, items, mp_lock, algorithm=None)

Bases: InMemoryBucket

classmethod init(rates, algorithm=None)

Creates a single ListProxy so that this bucket can be shared across multiple processes.

items
leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int

limiter_lock()

An additional lock to be used by Limiter in-front of the thread lock. Intended for multiprocessing environments where a thread lock is insufficient.

mp_lock
put(item)

Put an item (typically the current time) in the bucket return true if successful, otherwise false

Return type:

bool

class pyrate_limiter.buckets.MultiprocessStateStore(values, lock)

Bases: StateStore

State in a Manager list, guarded by a cross-process lock.

check(algorithm, rates, now, weight)

Apply algorithm.step to the stored state, atomically.

classmethod init()
Return type:

MultiprocessStateStore

is_async = False

None means “ask the Leaker to probe” (a client that may be either).

read(algorithm, rates)

Current state. For reporting only - never the basis of a decision.

reset()

Forget everything, as though the key had never been used.

Return type:

None

pyrate_limiter.buckets.PgQueries

alias of Queries

class pyrate_limiter.buckets.PostgresBucket(pool, table, rates, algorithm=None)

Bases: AbstractBucket

close()

Release any resources held by the bucket.

Subclasses may override this method to perform any necessary cleanup (e.g., closing files, network connections, or releasing locks) when the bucket is no longer needed.

count()

Count number of items in the bucket

Return type:

int | Awaitable[int]

flush()

Flush the whole bucket - Must remove failing-rate after flushing

Return type:

None | Awaitable[None]

is_async = False
leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int | Awaitable[int]

peek(index)

Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

Return type:

RateItem | None | Awaitable[RateItem | None]

pool
put(item)

Put an item (typically the current time) in the bucket return true if successful, otherwise false

Return type:

bool | Awaitable[bool]

table
class pyrate_limiter.buckets.RedisBucket(rates, redis, bucket_key, script_hash, algorithm=None)

Bases: AbstractBucket

A bucket using redis for storing data - We are not using redis’ built-in TIME since it is non-deterministic - In distributed context, use local server time or a remote time server - Each bucket instance use a dedicated connection to avoid race-condition - can be either sync or async

bucket_key
count()

Count number of items in the bucket

flush()

Flush the whole bucket - Must remove failing-rate after flushing

classmethod init(rates, redis, bucket_key, algorithm=None)
leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int | Awaitable[int]

now()

Retrieve current timestamp from the clock backend.

peek(index)

Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

Return type:

RateItem | None | Awaitable[RateItem | None]

put(item)

Add item to key

Return type:

bool | Awaitable[bool]

redis
script_hash
class pyrate_limiter.buckets.RedisStateStore(redis, key, ttl_ms=None)

Bases: StateStore

One Redis hash per key, holding a few floats however much traffic passes.

This is where the constant-state algorithms pay off: a sorted-set log grows with every consumed unit and must be trimmed, while this stays the same size and expires on its own.

The transition runs as Lua so the read-modify-write is atomic across clients. Works with either a sync or an async redis client.

check(algorithm, rates, now, weight)

Apply algorithm.step to the stored state, atomically.

default_clock = <pyrate_limiter.clocks.WallClock object>

State is shared between machines, where a monotonic clock is meaningless.

is_async = None

Unknown until the client is seen; the Leaker probes. leak() on StateBucket is a sync no-op either way.

read(algorithm, rates)

Current state. For reporting only - never the basis of a decision.

reset()

Forget everything, as though the key had never been used.

Return type:

None | Awaitable[None]

class pyrate_limiter.buckets.SQLiteBucket(rates, conn, table, lock=None, algorithm=None)

Bases: AbstractBucket

For sqlite bucket, we are using the sql time function as the clock item’s timestamp wont matter here

close()

Release any resources held by the bucket.

Subclasses may override this method to perform any necessary cleanup (e.g., closing files, network connections, or releasing locks) when the bucket is no longer needed.

conn
count()

Count number of items in the bucket

Return type:

int

flush()

Flush the whole bucket - Must remove failing-rate after flushing

Return type:

None

full_count_query
classmethod init_from_file(rates, table='rate_bucket', db_path=None, create_new_table=True, use_file_lock=False, algorithm=None)
Return type:

SQLiteBucket

is_async = False
leak(current_timestamp=None)

Leaking/clean up bucket

Return type:

int

limiter_lock()

An additional lock to be used by Limiter in-front of the thread lock. Intended for multiprocessing environments where a thread lock is insufficient.

lock
now()

Retrieve current timestamp from the clock backend.

peek(index)

Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

Return type:

RateItem | None

put(item)

Put an item (typically the current time) in the bucket return true if successful, otherwise false

Return type:

bool

table
use_limiter_lock
class pyrate_limiter.buckets.SQLiteClock(conn)

Bases: AbstractClock

Get timestamp using SQLite as remote clock backend

__init__(conn)

In multiprocessing cases, use the bucket, so that a shared lock is used.

classmethod default()
lock
now()

Get time as of now, in milliseconds

Return type:

int

time_query = "SELECT CAST(ROUND((julianday('now') - 2440587.5)*86400000) As INTEGER)"
pyrate_limiter.buckets.SQLiteQueries

alias of Queries

class pyrate_limiter.buckets.StateBucket(rates, algorithm=None, store=None, clock=None)

Bases: AbstractBucket

Bucket for constant-state algorithms - GCRA, TokenBucket.

Keeps a few numbers per key rather than an entry per consumed unit, so storage does not grow with traffic and the wait is exact without a lookup.

The log contract does not apply: peek() has nothing to return and leak() nothing to trim. Use count() for how many units are currently owed.

algorithm
close()

Release any resources held by the bucket.

Subclasses may override this method to perform any necessary cleanup (e.g., closing files, network connections, or releasing locks) when the bucket is no longer needed.

Return type:

None

count()

Units currently owed to the bucket - an estimate, not a log length.

Return type:

int | Awaitable[int]

flush()

Flush the whole bucket - Must remove failing-rate after flushing

Return type:

None | Awaitable[None]

is_async = False
leak(current_timestamp=None)

No-op: state is constant-size, so there is nothing to trim.

Shared stores expire idle keys themselves (Redis via a TTL).

Return type:

int

peek(index)

Always None: this bucket keeps no per-item log to peek into.

Return type:

RateItem | None

put(item)

Put an item (typically the current time) in the bucket return true if successful, otherwise false

Return type:

bool | Awaitable[bool]

store
waiting(item)

Wait recorded by the last put(), or re-derived for a different weight.

Never inspects a log the way the window buckets do - there is none. When the query does not match the last put, the wait is recomputed by replaying step() against the stored state, which spends nothing.

Return type:

int | Awaitable[int]

Submodules