Rate Limiter

A Rate Limiter caps how many requests a client can make inside a time window. Use it to protect a service from overload and to enforce fair usage across clients.

Why

Protect services from overload and abuse.
Enforce fair usage across clients.
Produce HTTP 429 responses with RFC 9211 RateLimit-* or legacy X-RateLimit-* headers.

RateLimiter is algorithm-agnostic. Pass an algorithm config to choose semantics. Everything else (API, RateLimitResult, backend registry, fail_open) is shared.

Usage

Load a backend, build a limiter with a factory classmethod, then call acquire:

from grelmicro.resilience import RateLimiter

# Sliding window for precise API throttling.
auth_limiter = RateLimiter.sliding_window("auth", limit=5, window=60)

# Token bucket for burst-friendly "N then 1/sec" semantics.
api_limiter = RateLimiter.token_bucket("api", capacity=100, refill_rate=10)


async def login(ip: str) -> None:
    result = await auth_limiter.acquire(key=ip)
    if not result.allowed:
        print(f"Too many attempts, retry after {result.retry_after:.0f}s")
        return
    print(f"Login allowed, {result.remaining} attempts remaining")


async def api_call(user_id: str) -> None:
    # Raises RateLimitExceededError if the bucket is empty
    await api_limiter.acquire_or_raise(key=user_id)
    print("API call allowed")

Checking the limit

Pick the call by what you need:

Need	Method	Why
Just branch yes or no	`allow()`	Smallest code.
Build HTTP 429 headers	`acquire()`	Keeps `retry_after`, `remaining`, and `reset_after`.
Let a shared handler map rejections	`acquire_or_raise()`	Raises `RateLimitExceededError`, an `AdmissionError`.
Smooth work instead of rejecting	`wait()`	Sleeps until admitted, with optional `max_wait`.

The simplest form is a boolean:

if await limiter.allow(key="user-1"):
    ...  # served
else:
    ...  # throttled

acquire returns the full RateLimitResult when you need the metadata. It reads as a boolean too, so you branch on it directly and still keep retry_after/remaining on the deny side:

result = await limiter.acquire(key="user-1")
if not result:
    # reject with a 429 and a Retry-After of result.retry_after seconds
    ...

Use acquire_or_raise when a surrounding layer should turn the rejection into a response: it raises RateLimitExceededError, which is an AdmissionError (the shared base, exported from the top-level grelmicro package for every "turned away" rejection: rate limiter, bulkhead, open circuit breaker, non-blocking lock), so one except AdmissionError catches them all.

Waiting until allowed

wait blocks until tokens are available, then consumes them. Use it to smooth a burst into the limit instead of rejecting it, for example when calling a rate-limited upstream:

from grelmicro.resilience import RateLimiter

api_limiter = RateLimiter.token_bucket("api", capacity=100, refill_rate=10)


async def serve(user_id: str) -> None:
    # Block until a token frees up, then proceed.
    await api_limiter.wait(key=user_id)


async def serve_bounded(user_id: str) -> None:
    # Give up after 2 seconds, raising RateLimitExceededError.
    await api_limiter.wait(key=user_id, cost=3, max_wait=2.0)

It polls acquire on the clock seam, sleeping retry_after between attempts, so a denied call never consumes tokens. By default it waits as long as needed. Pass max_wait to bound the wait: it raises RateLimitExceededError once the budget would be exceeded. A cost larger than the limit raises ValueError instead of waiting forever.

One fleet-wide limit

When the limiter protects a service with one shared budget (no per-user or per-IP split), omit key. It defaults to "default", and the limiter's own name already namespaces the bucket on the backend:

api_limiter = RateLimiter.token_bucket("api", capacity=5, refill_rate=1)

await api_limiter.acquire()             # one fleet-wide bucket
await api_limiter.allow()
await api_limiter.acquire_or_raise()

await api_limiter.acquire(key=user_id)  # per-subject stays explicit

The factory classmethods keep the call site explicit and short:

from grelmicro.resilience import RateLimiter

auth_limiter = RateLimiter.sliding_window("auth", limit=5, window=60)
api_limiter = RateLimiter.token_bucket(
    "api",
    capacity=100,
    refill_rate=10,
)

Advanced

For the from_config declarative path and pydantic-settings composition, see Declarative configuration.

RateLimiter intentionally does not flatten both algorithms into one generic kwargs constructor. Token bucket and sliding window have different parameter vocabularies, and keeping one explicit entry point per behaviour makes the public API easier to read.

Choosing an algorithm

Pick the algorithm whose behaviour matches how operators describe the limit in runbooks and API docs. Both algorithms share the same Python API, backends, and RateLimitResult shape, so you can switch later.

Throttling an HTTP API with RateLimit-* headers? Use SlidingWindowConfig.
Want "allow a burst of N, then 1 per second sustained"? Use TokenBucketConfig.

Decision guide, side-by-side, and worked scenarios

Decision guide

Are you throttling an HTTP API with RateLimit-* or X-RateLimit-* headers? Use SlidingWindowConfig. It matches the IETF RateLimit headers directly and produces precise limit, remaining, and reset_after values.
Do you want "allow a burst of N, then 1 per second sustained"? Use TokenBucketConfig. The capacity and refill_rate parameters describe exactly that.
Does a client need to send occasional spikes above the average rate? Use TokenBucketConfig. The capacity absorbs the spike.
Did you search for "leaky bucket"? Use SlidingWindowConfig. It is the leaky-bucket-as-meter formulation.

Side-by-side

	SlidingWindowConfig	TokenBucketConfig
Mental model	"N requests per sliding T-second window"	"A bucket holding N tokens that refills at R tokens/sec"
Parameters	`limit`, `window`	`capacity`, `refill_rate`
Burst behaviour	Up to `limit` requests if the window is empty	Up to `capacity` if the bucket is full
Sustained rate	`limit / window` requests per second	`refill_rate` tokens per second
HTTP header fit	Strong. `reset_after` is a true window boundary and maps directly to `RateLimit-Reset`.	Workable. `retry_after` is the time until the next token (continuous refill), not a window reset.

Worked scenarios

"Limit each user to 100 API calls per minute." Use SlidingWindowConfig(limit=100, window=60). The sliding window matches the natural description, and RateLimitResult.reset_after feeds directly into RateLimit-Reset.
"Allow a burst of 20 uploads, then 2 per second." Use TokenBucketConfig(capacity=20, refill_rate=2). Each word in the sentence maps to one parameter.
"Fair share. Every account gets 1 heavy job per 10 seconds but can queue up to 5." Use TokenBucketConfig(capacity=5, refill_rate=0.1).
"Throttle expensive webhook retries. At most 10 per minute per target." Use SlidingWindowConfig(limit=10, window=60).

There is no separate LeakyBucket algorithm. SlidingWindowConfig is the leaky-bucket-as-meter formulation. Operators searching for "leaky bucket" should use SlidingWindowConfig.

Performance

Both algorithms run in O(1) per operation. End-to-end latency is dominated by the backend: a Redis round-trip costs far more than the algorithm itself. Per-key memory on the Memory backend differs by about 15 MB per million keys. Choose based on behaviour, not compute cost.

Backend

Load a backend before using RateLimiter. The same backend serves every algorithm.

Install

The Redis backend needs the redis extra, the Postgres backend needs the postgres extra, and the SQLite backend needs the sqlite extra: pip install "grelmicro[redis]", pip install "grelmicro[postgres]", or pip install "grelmicro[sqlite]". See the installation guide for uv and poetry.

RedisPostgresSQLiteMemory

from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import RateLimiterRegistry

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[RateLimiterRegistry(redis)])

from grelmicro import Grelmicro
from grelmicro.providers.postgres import PostgresProvider
from grelmicro.resilience import RateLimiterRegistry

postgres = PostgresProvider("postgresql://localhost:5432/app")
micro = Grelmicro(uses=[RateLimiterRegistry(postgres)])

from grelmicro import Grelmicro
from grelmicro.providers.sqlite import SQLiteProvider
from grelmicro.resilience import RateLimiterRegistry

sqlite = SQLiteProvider("rate_limit.db")
micro = Grelmicro(uses=[sqlite, RateLimiterRegistry(sqlite)])

from grelmicro.resilience.ratelimiter.memory import MemoryRateLimiterAdapter

backend = MemoryRateLimiterAdapter()

Warning

Please make sure to use a proper way to store connection URLs, such as environment variables, not hard-coded strings like the example above.

	Redis	Postgres	SQLite	Memory
Use case	Production	Production (when Postgres is already deployed)	Single host that needs durability	Testing / single-process
Multi-node	Yes	Yes	No	No
Persistence	Yes (auto-expiring keys)	Yes (table-backed)	Yes (file-backed)	No

Choosing a backend

Use Redis in production when you already run Redis and want the lowest-latency distributed limiter. Use Postgres when Postgres is your only stateful dependency and you want one fewer service to run. Use SQLite for a single host that needs limits to survive restarts. Use Memory for tests and single-process apps. Redis and Postgres coordinate across replicas. SQLite and Memory do not.

How each backend stores state

The Postgres adapter stores state in a single grelmicro_rate_limiter table. acquire and peek each run one round-trip to a PL/pgSQL function. Concurrent writes for the same key are serialized with pg_advisory_xact_lock. reset clears the key with a plain DELETE. The table and functions are created on first connect: pass auto_migrate=False when your own migration tool owns the schema.

SQLite uses a SQLiteProvider, the same provider-first shape as Redis and Postgres. Pass the path to the provider or set the SQLITE_PATH environment variable. State lives in a single grelmicro_rate_limiter table in the file. Each acquire runs a read-modify-write inside a BEGIN IMMEDIATE transaction. The provider's lock serializes the single connection within the process, and the transaction's write lock serializes across processes sharing the file. Use it for a single host that wants durability without running a separate service.

The backend compiles the algorithm into a bound strategy at RateLimiter.__init__ through backend.bind(config). Runtime acquire, peek, and reset calls invoke that strategy directly. There is no algorithm dispatch on the request path.

Tip

The rate limiter uses the same backend registry pattern as the synchronization primitives. See Backend Architecture for details.

Coming from 0.x: register a backend, then install the app

In 0.x you opened a global backend in the lifespan (async with RedisRateLimiterBackend(...)). In 1.0 you register a RateLimiterRegistry on the app and wire the app with micro.install(app):

micro = Grelmicro(uses=[RateLimiterRegistry(RedisRateLimiterAdapter())])
micro.install(app)  # opens the registry AND binds it per request

install is the important part. A module-level RateLimiter("auth") resolves its backend from the active app per request, which only works when install adds its middleware. Open async with micro: in a hand-written lifespan without install and the app starts up healthy, then raises OutOfContextError on the first rate-limited request. See Wiring an App for the guard and the micro.check_ambient_binding(app) test helper.

Result fields

RateLimitResult is the same across algorithms and carries everything needed for HTTP rate limit headers. The HTTP header column shows the RFC 9211 name first and the legacy X-RateLimit-* name second. Pick whichever convention your API already uses.

Field	Type	Description	HTTP Header
`allowed`	`bool`	Whether the request is permitted	200 vs 429 status
`limit`	`int`	Total quota (`limit` for SlidingWindowConfig, `int(capacity)` for TokenBucketConfig)	`RateLimit-Limit` / `X-RateLimit-Limit`
`remaining`	`int`	Remaining requests / tokens	`RateLimit-Remaining` / `X-RateLimit-Remaining`
`retry_after`	`float`	Seconds until next allowed request	`Retry-After`
`reset_after`	`float`	Seconds until full quota resets	`RateLimit-Reset` / `X-RateLimit-Reset`

Weighted requests

Use the cost parameter to consume multiple tokens per request.

# Bulk endpoint costs 10 tokens
result = await api_limiter.acquire(key=user_id, cost=10)

Peek (check without consuming)

Use peek() to inspect current state without consuming tokens.

from grelmicro.resilience import RateLimiter

invite_limiter = RateLimiter.sliding_window("invite", limit=5, window=3600)


async def is_locked(code: str) -> bool:
    result = await invite_limiter.peek(key=code)
    return not result.allowed

Reset

Use reset() to delete the state for a key, restoring its full quota.

from grelmicro.resilience import RateLimiter

auth_limiter = RateLimiter.sliding_window("auth", limit=5, window=60)


def verify_password(password: str) -> bool:
    return True


async def login(ip: str, password: str) -> None:
    await auth_limiter.acquire_or_raise(key=ip)

    if verify_password(password):
        # Successful login: clear the failure counter
        await auth_limiter.reset(key=ip)

Fail-open mode

Use fail_open=True when availability matters more than strictness. On backend errors (e.g. Redis down), the rate limiter returns an allowed result instead of raising.

from grelmicro.resilience import RateLimiter

# Non-critical limiter: prefer availability over strictness
limiter = RateLimiter.token_bucket(
    "analytics",
    capacity=100,
    refill_rate=10,
    fail_open=True,
)


def record_event(user_id: str) -> None: ...


async def track_event(user_id: str) -> None:
    # If Redis is down, the event is still tracked
    result = await limiter.acquire(key=user_id)
    if result.allowed:
        record_event(user_id)

Warning

Fail-open mode only catches backend infrastructure errors. Legitimate rate-limit rejections still work normally.

Standalone `MemoryTokenBucket`

MemoryTokenBucket is a standalone, synchronous, thread-safe in-memory token-bucket primitive. Unlike RateLimiter, it is not pluggable and not async. Use it when you need a raw, zero-I/O bucket on a synchronous performance-critical path. It powers grelmicro.log.RateLimitFilter, which is the recommended way to use it for rate-limiting log records. Call it directly for any other use case.

Usage

from grelmicro.resilience import MemoryTokenBucket

# Sync, thread-safe, zero-I/O primitive.
# Useful for CLI tools, shell helpers, and other sync hot paths
# where the async RateLimiter isn't appropriate.
bucket = MemoryTokenBucket(capacity=5, refill_rate=1)


def handle_event(event_id: str) -> None:
    if not bucket.try_acquire(key=event_id):
        return
    # ... process the event

API

Method	Description
`try_acquire(key="", *, cost=1.0) -> bool`	Consume `cost` tokens and return `True` if allowed.
`peek(key="") -> float`	Current token count (fractional).
`reset(key="") -> None`	Restore `key` to full capacity.
`capacity` / `refill_rate`	Read-only configuration.

Rate Limiter

Usage

Checking the limit

Waiting until allowed

One fleet-wide limit

Choosing an algorithm

Decision guide

Side-by-side

Worked scenarios

Backend

Choosing a backend

Result fields

Weighted requests

Peek (check without consuming)

Reset

Fail-open mode

Standalone MemoryTokenBucket

Usage

API

Standalone `MemoryTokenBucket`