pyrate_limiter package

class pyrate_limiter.AbstractBucket

Bases: ABC

Base bucket interface Assumption: len(rates) always > 0 TODO: allow empty rates

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

abstractmethod count()

Count number of items in the bucket

Return type:

int | Awaitable[int]

failing_rate = None
abstractmethod flush()

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

Return type:

None | Awaitable[None]

is_async = None
abstractmethod leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int | Awaitable[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.

Return type:

object | None

now()

Retrieve current timestamp from the clock backend.

abstractmethod 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]

abstractmethod put(item)

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

Return type:

bool | Awaitable[bool]

put_decision(item)

put(), returning the full Decision rather than a bare bool.

Buckets need not override it; the default reads back what put() recorded. retry_after_ms is None for buckets that record none.

Return type:

Decision | Awaitable[Decision]

property rates
waiting(item)

Calculate time until bucket become availabe to consume an item again

Return type:

int | Awaitable[int]

class pyrate_limiter.AbstractClock

Bases: ABC

Clock that return timestamp for now

abstractmethod now()

Get time as of now, in milliseconds

Return type:

int | Awaitable[int]

class pyrate_limiter.Algorithm

Bases: ABC

A rate-limiting policy, independent of any storage backend.

Implementations must be stateless so one instance can be shared across buckets and threads. The two sub-interfaces differ in what they need remembered per key: LogAlgorithm an entry per consumed unit, StateAlgorithm a fixed handful of numbers.

max_weight(rate)

Largest weight this policy can ever admit under rate.

Return type:

int

class pyrate_limiter.BucketAsyncWrapper(bucket)

Bases: AbstractBucket

BucketAsyncWrapper is a wrapping over any bucket that turns a async/synchronous bucket into an async one

async count()

Count number of items in the bucket

property failing_rate

The type of the None singleton.

async flush()

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

Return type:

None

is_async = True
async leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int

now()

Retrieve current timestamp from the clock backend.

Return type:

int

async 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

async put(item)

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

property rates
async waiting(item)

Calculate time until bucket become availabe to consume an item again

Return type:

int

class pyrate_limiter.BucketFactory

Bases: ABC

Asbtract BucketFactory class. It is reserved for user to implement/override this class with his own bucket-routing/creating logic

close()
Return type:

None

create(bucket_class, *args, **kwargs)

Creating a bucket dynamically

Return type:

AbstractBucket

dispose(bucket)

Delete a bucket from the factory

Return type:

bool

abstractmethod get(item)

Get the corresponding bucket to this item

Return type:

AbstractBucket | Awaitable[AbstractBucket]

get_buckets()

Iterator over all buckets in the factory

Return type:

List[AbstractBucket]

property leak_interval

Retrieve leak-interval from inner Leaker task

schedule_leak(new_bucket)

Schedule all the buckets’ leak, reset bucket’s failing rate

Return type:

None

abstractmethod wrap_item(name, weight=1)

Add the current timestamp to the receiving item using any clock backend - Turn it into a RateItem - Can return either a coroutine or a RateItem instance

Return type:

RateItem | Awaitable[RateItem]

class pyrate_limiter.Decision(failing_rate=None, retry_after_ms=None)

Bases: object

Outcome of an admit check.

retry_after_ms is measured from the checked item’s own timestamp. None means “unknown, ask AbstractBucket.waiting()” - either the weight can never fit, or the backend does not compute a wait. It does not mean “no wait”.

property allowed
failing_rate = None
retry_after_ms = None
class pyrate_limiter.Duration(*values)

Bases: Enum

Interval helper class

DAY = 86400000
HOUR = 3600000
MINUTE = 60000
SECOND = 1000
WEEK = 604800000
static readable(value)
Return type:

str

class pyrate_limiter.FixedWindow

Bases: LogAlgorithm

Counts within a wall-clock-aligned window that resets every interval.

Cheaper and coarser than the rolling window: up to 2 * limit can pass across a window boundary. Use it to mirror an upstream API that genuinely resets on the hour rather than rolling.

admit(rates, counts, weight)

Whether weight more units fit, given counts aligned to rates.

Return type:

Decision

retry_after(rate, now, blocking_timestamp)

Milliseconds until room exists under rate.

blocking_timestamp is the entry named by blocking_offset(), or None when there is none - or when the policy never asks for one.

Return type:

int

window_start(rate, now)

Inclusive lower bound of rate’s counting window at now.

Return type:

int

class pyrate_limiter.GCRA

Bases: StateAlgorithm

Generic Cell Rate Algorithm - a leaky bucket kept as one timestamp.

Tracks a theoretical arrival time (TAT) per rate: the moment the bucket would next be empty. Admitting weight pushes the TAT forward by weight * emission_interval; the request is allowed while that stays within burst units of now.

Sustains limit per interval while tolerating a burst of rate.burst, using one number per rate instead of an entry per unit.

State is integer microseconds, not fractional milliseconds. An absolute TAT in epoch ms is ~1.7e12, and accumulating a fractional emission interval onto it loses the low bits - enough that the accumulated sum of burst emissions no longer equals burst * emission, and the last unit of a full burst gets rejected by a rounding error. Integers make it exact, and stay well inside the 2**53 a Lua double holds.

consumed(rates, state, now)

Units currently owed - the closest analogue to a log’s length.

Return type:

int

decode(values)

Parse persisted strings back into state.

Return type:

Tuple[float, ...]

initial(rates)

State for a key that has never been used.

Return type:

Tuple[float, ...]

max_weight(rate)

Largest weight this policy can ever admit under rate.

Return type:

int

redis_args(rates)

Arguments redis_script() needs, after the standard header.

The store passes these through without inspecting them, so a policy’s script and its arguments stay a matched pair that only the policy knows the shape of. The header the store supplies first is now, weight, ttl_ms, len(rates).

Return type:

List[int | float]

redis_script()

Lua implementing step() atomically, if this policy has one.

Return type:

str | None

step(rates, state, now, weight)

Apply an arrival of weight at now.

Returns the state to persist and the verdict. On denial it must return state unchanged: a rejected request spends nothing, under any rate.

Return type:

Tuple[Tuple[float, ...], Decision]

class pyrate_limiter.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.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.Limiter(argument, buffer_ms=50)

Bases: object

This class responsibility is to sum up all underlying logic and make working with async/sync functions easily

__init__(argument, buffer_ms=50)

Init Limiter using either a single bucket / multiple-bucket factory / single rate / rate list.

Parameters:

argument (BucketFactory | AbstractBucket | Rate | List[Rate]) – The bucket or rate configuration.

as_decorator(*, name='ratelimiter', weight=1)
bucket_factory
buckets()

Get list of active buckets

Return type:

List[AbstractBucket]

buffer_ms
close()
Return type:

None

dispose(bucket)

Dispose/Remove a specific bucket, using bucket-id or bucket object as param

Return type:

bool

handle_bucket_put(bucket, item, blocking, _force_async=False, deadline=None)

Putting item into bucket

Return type:

bool | Awaitable[bool]

lock
try_acquire(name='pyrate', weight=1, blocking=True, timeout=-1)

Attempt to acquire a permit from the limiter.

Parameters:

  • name (str) – The bucket key to acquire from.

  • weight (int) – Number of permits to consume.

  • timeout (int | float) – Maximum time (in seconds) to wait; -1 means wait indefinitely.

  • blocking (bool) – If True, block until a permit is available (subject to timeout); if False, return immediately.

Returns:

True if the permit was acquired, False otherwise. Async limiters return an awaitable resolving to the same.

Return type:

bool | Awaitable[bool]

async try_acquire_async(name='pyrate', weight=1, blocking=True, timeout=-1)

Attempt to asynchronously acquire a permit from the limiter.

Parameters:

  • name (str) – The bucket key to acquire from.

  • weight (int) – Number of permits to consume.

  • blocking (bool) – If True, wait until a permit is available (subject to timeout); if False, return immediately.

  • timeout (int | float) – Maximum time (in seconds) to wait; -1 means wait indefinitely.

Returns:

True if the permit was acquired, False otherwise.

Return type:

bool

Notes

This is the async variant of try_acquire. A top-level, thread-local async lock is used to prevent blocking the event loop.

class pyrate_limiter.LogAlgorithm

Bases: Algorithm

Policy over storage holding one timestamped entry per consumed unit.

abstractmethod admit(rates, counts, weight)

Whether weight more units fit, given counts aligned to rates.

Return type:

Decision

blocking_offset(rate, weight)

Offset from the newest stored entry (0-based) whose expiry makes room for weight, or None if the wait does not depend on an entry.

Return type:

int | None

decide(rates, counts, weight, now, peek_timestamp)

admit(), resolving the retry-after in the same step on denial.

peek_timestamp(offset) is only called when the policy asks for an entry and the item was rejected, so backends pay for the lookup only when it is needed.

Return type:

Decision

leak_bound(rates, now)

Timestamp below which an entry is outside every rate’s window.

Return type:

int

abstractmethod retry_after(rate, now, blocking_timestamp)

Milliseconds until room exists under rate.

blocking_timestamp is the entry named by blocking_offset(), or None when there is none - or when the policy never asks for one.

Return type:

int

abstractmethod window_start(rate, now)

Inclusive lower bound of rate’s counting window at now.

Return type:

int

class pyrate_limiter.MonotonicAsyncClock

Bases: AbstractClock

Monotonic Async Clock, meant for testing only

async now()

Get monotonic time in milliseconds

Return type:

int

class pyrate_limiter.MonotonicClock

Bases: AbstractClock

now()

Get monotonic time in milliseconds

Return type:

int

class pyrate_limiter.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.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.PgQueries

alias of Queries

class pyrate_limiter.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.PostgresClock(pool)

Bases: AbstractClock

Get timestamp using Postgres as remote clock backend

now()

Get current time in milliseconds using Postgres.

Falls back to local time if the DB query fails for any reason.

Return type:

int

class pyrate_limiter.Rate(limit, interval, burst=None)

Bases: object

Rate definition.

Parameters:

  • limit (int) – Number of requests allowed within interval

  • interval (int | Duration) – Time interval, in miliseconds

  • burst (int | None) – How many units may be spent at once. Only the constant-state algorithms (GCRA, TokenBucket) read it; the window algorithms admit up to limit per window regardless. Defaults to limit, which is classic token-bucket behaviour - a full bucket at rest. burst=1 makes the output perfectly smooth.

burst
interval
limit
class pyrate_limiter.RateItem(name, timestamp, weight=1)

Bases: object

RateItem is a wrapper for bucket to work with

name
timestamp
weight
class pyrate_limiter.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.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.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.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.SQLiteQueries

alias of Queries

class pyrate_limiter.SingleBucketFactory(bucket, schedule_leak=True)

Bases: BucketFactory

Single-bucket factory for quick use with Limiter

__init__(bucket, schedule_leak=True)

Initialize the SingleBucketFactory with a bucket and an optional leak scheduling flag.

schedule_leak (bool): If True, the factory will schedule periodic leaks for the bucket. Default is True. Disable only if you plan to handle leaking manually.

bucket
get(_)

Get the corresponding bucket to this item

Return type:

AbstractBucket

wrap_item(name, weight=1)

Add the current timestamp to the receiving item using any clock backend - Turn it into a RateItem - Can return either a coroutine or a RateItem instance

class pyrate_limiter.SlidingWindowLog

Bases: LogAlgorithm

Precise rolling window: admit while each rate’s last interval stays under its limit.

The default. Exact, at the cost of one stored entry per consumed unit.

admit(rates, counts, weight)

Whether weight more units fit, given counts aligned to rates.

Return type:

Decision

blocking_offset(rate, weight)

Offset from the newest stored entry (0-based) whose expiry makes room for weight, or None if the wait does not depend on an entry.

Return type:

int | None

retry_after(rate, now, blocking_timestamp)

Milliseconds until room exists under rate.

blocking_timestamp is the entry named by blocking_offset(), or None when there is none - or when the policy never asks for one.

Return type:

int

window_start(rate, now)

Inclusive lower bound of rate’s counting window at now.

Return type:

int

class pyrate_limiter.StateAlgorithm

Bases: Algorithm

Policy whose state is a fixed-size tuple of numbers, not a log.

Storage keeps one small value per key however much traffic passes, and the wait comes out in closed form. In exchange the check is destructive - it spends what it admits - so step() must evaluate every rate before committing any of them.

consumed(rates, state, now)

Units currently owed - the closest analogue to a log’s length.

Return type:

int

decode(values)

Parse persisted strings back into state.

Return type:

Tuple[float, ...]

abstractmethod initial(rates)

State for a key that has never been used.

Return type:

Tuple[float, ...]

redis_args(rates)

Arguments redis_script() needs, after the standard header.

The store passes these through without inspecting them, so a policy’s script and its arguments stay a matched pair that only the policy knows the shape of. The header the store supplies first is now, weight, ttl_ms, len(rates).

Return type:

List[int | float]

redis_script()

Lua implementing step() atomically, if this policy has one.

Return type:

str | None

abstractmethod step(rates, state, now, weight)

Apply an arrival of weight at now.

Returns the state to persist and the verdict. On denial it must return state unchanged: a rejected request spends nothing, under any rate.

Return type:

Tuple[Tuple[float, ...], Decision]

class pyrate_limiter.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]

class pyrate_limiter.StateStore

Bases: ABC

Holds one key’s state for a StateAlgorithm.

The store’s only real job is atomicity: check() must read the state, apply the transition and persist the result without another writer interleaving. How it achieves that is its own business - a lock in-process, a Lua script in Redis.

abstractmethod check(algorithm, rates, now, weight)

Apply algorithm.step to the stored state, atomically.

Return type:

Decision | Awaitable[Decision]

close()

Release any resources held. Optional.

Return type:

None

default_clock = <pyrate_limiter.clocks.MonotonicClock object>

Used when the bucket is not given a clock. Shared stores override it, since a monotonic clock means nothing across machines.

is_async = False

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

abstractmethod read(algorithm, rates)

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

Return type:

Tuple[float, ...] | Awaitable[Tuple[float, ...]]

abstractmethod reset()

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

Return type:

None | Awaitable[None]

class pyrate_limiter.TokenBucket

Bases: GCRA

Token bucket, which is GCRA under a more familiar name.

A bucket of rate.burst tokens refilling at rate.limit / rate.interval admits exactly what GCRA does with an emission interval of interval / limit. Same implementation, one float of state rather than a token count plus a refill timestamp.

class pyrate_limiter.WallClock

Bases: AbstractClock

Wall-clock epoch milliseconds.

Needed where timestamps are compared across processes or hosts - a monotonic clock is only meaningful within one machine’s boot. Use it for any bucket whose state is shared through Redis.

now()

Get time as of now, in milliseconds

Return type:

int

pyrate_limiter.dedicated_sqlite_clock_connection()
pyrate_limiter.id_generator(size=10)
Return type:

str

pyrate_limiter.validate_rate_list(rates)

Raise false if rates are incorrectly ordered.

Return type:

bool

Subpackages

Submodules