orionis.http.layer.store.memory_rate_limit.MemoryRateLimitStore

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class MemoryRateLimitStore: (source)

Constructor: MemoryRateLimitStore(max_keys, max_events)

View In Hierarchy

Undocumented

Method	`__atCapacity`	Report whether admitting a new client would exceed either bound.
Method	`__gc`	Inspect the next group of keys and remove expired buckets.
Method	`__hit`	Update quota and capacity counters while holding the store lock.
Method	`__init__`	Initialize an empty rate-limit store.
Method	`__newBucket`	Create a bucket only when bounded reclamation makes room for it.
Async Method	`hit`	Record a request attempt and decide whether it is allowed.
Constant	`_GC_BATCH_SIZE`	Undocumented
Constant	`_GC_INTERVAL`	Undocumented
Class Variable	`__slots__`	Undocumented
Instance Variable	`__keys`	Undocumented
Instance Variable	`__lock`	Undocumented
Instance Variable	`__max_events`	Undocumented
Instance Variable	`__max_keys`	Undocumented
Instance Variable	`__storage`	Undocumented
Instance Variable	`__ticks`	Undocumented
Instance Variable	`__total_events`	Undocumented

def __atCapacity(self) -> bool: (source) ¶

Report whether admitting a new client would exceed either bound.

Returns
`bool`	Whether client or timestamp capacity is exhausted.

def __gc(self, now: float): (source) ¶

Inspect the next group of keys and remove expired buckets.

Parameters
now:`float`	Current monotonic timestamp.

Returns
`None`	Undocumented

def __hit(self, key: str, limit: int, window: int) -> bool: (source) ¶

Update quota and capacity counters while holding the store lock.

Parameters
key:`str`	Client identity whose timestamps are retained.
limit:`int`	Maximum accepted attempts per client window.
window:`int`	Sliding-window duration in seconds.

Returns
`bool`	Whether quota and retained-object capacity permit this attempt.

def __init__(self, *, max_keys: int = 10000, max_events: int = 100000): (source) ¶

Initialize an empty rate-limit store.

Parameters
max_keys:`int, optional`	Maximum number of retained client buckets.
max_events:`int, optional`	Maximum number of accepted timestamps retained across all clients.

Returns
`None`	Undocumented

def __newBucket(self, key: str, now: float, window: int) -> _RateLimitBucket | None: (source) ¶

Create a bucket only when bounded reclamation makes room for it.

Parameters
key:`str`	Client identity to retain.
now:`float`	Current monotonic timestamp.
window:`int`	Sliding-window duration in seconds.

Returns
`_RateLimitBucket \| None`	New empty bucket, or None when the store cannot retain more clients.

async def hit(self, key: str, limit: int, window: int) -> bool: (source) ¶

Record a request attempt and decide whether it is allowed.

Implements a sliding window over accepted attempts. Inactive keys are reclaimed incrementally during subsequent attempts, including rejected attempts. Each key should use a consistent window. A lock serializes updates across event loops and threads. Capacity exhaustion rejects attempts without evicting active quotas or retaining rejected clients.

Parameters
key:`str`	Unique identifier for the rate-limited entity (e.g. IP, user id, route).
limit:`int`	Maximum number of requests allowed within `window`.
window:`int`	Length of the sliding window in seconds.