Resilience

grelmicro.resilience

Resilience.

RateLimiterConfig module-attribute

RateLimiterConfig = TokenBucketConfig | GCRAConfig

Discriminated union of supported rate-limiter algorithm configurations.

CircuitBreaker

CircuitBreaker(
    name: str,
    *,
    ignore_exceptions: type[Exception]
    | str
    | tuple[type[Exception] | str, ...]
    | None = None,
    error_threshold: PositiveInt | None = None,
    success_threshold: PositiveInt | None = None,
    reset_timeout: PositiveFloat | None = None,
    half_open_capacity: PositiveInt | None = None,
    log_level: LogLevel | None = None,
    env_prefix: str | None = None,
    read_env: bool = True,
)

Circuit Breaker.

Implements the circuit breaker pattern. It watches calls to a protected service and blocks them when the service is failing, to avoid cascading errors.

Initialize the circuit breaker.

PARAMETER	DESCRIPTION
name	Name of the circuit breaker instance. Acts as the instance identity. Used as the env var prefix and exposed via the name property. TYPE: str
ignore_exceptions	Exceptions ignored by the breaker. Errors of these types do not count toward error_threshold. Accepts a single exception class, a tuple, or fully-qualified import strings such as "builtins.ValueError" or "my_app.errors.PaymentError". When unset, resolves from the env path or falls back to the CircuitBreakerConfig default (empty tuple). TYPE: type[Exception] | str | tuple[type[Exception] | str, ...] | None DEFAULT: None
error_threshold	Consecutive errors before the breaker opens. Default: 5. When unset, resolves from GREL_CIRCUIT_BREAKER_{NAME_UPPER}_ERROR_THRESHOLD if present, otherwise falls back to the CircuitBreakerConfig default. TYPE: PositiveInt | None DEFAULT: None
success_threshold	Consecutive successes in HALF_OPEN state before the breaker closes. Default: 2. TYPE: PositiveInt | None DEFAULT: None
reset_timeout	Seconds the breaker stays OPEN before transitioning to HALF_OPEN. Default: 30.0. TYPE: PositiveFloat | None DEFAULT: None
half_open_capacity	Maximum concurrent calls allowed in the HALF_OPEN state. Default: 1. TYPE: PositiveInt | None DEFAULT: None
log_level	Logging level for state-change messages. Default: WARNING. TYPE: LogLevel | None DEFAULT: None
env_prefix	Override the auto-derived environment variable prefix. Default: GREL_CIRCUIT_BREAKER_{NAME_UPPER}_. TYPE: str | None DEFAULT: None
read_env	Whether to read environment variables. Default: True. TYPE: bool DEFAULT: True

from_thread property

from_thread: _ThreadAdapter

Return the lock adapter for worker thread.

name property

name: str

Return the name of the circuit breaker.

state property

state: CircuitBreakerState

Return the current state of the circuit breaker.

last_error property

last_error: Exception | None

Return the last error recorded by the circuit breaker.

last_error_time property

last_error_time: datetime | None

Return the time of the last error recorded by the circuit breaker.

config property

config: CircuitBreakerConfig

Return the circuit breaker configuration.

from_config classmethod

from_config(
    name: str, config: CircuitBreakerConfig
) -> Self

Construct a CircuitBreaker from a name and a pre-built CircuitBreakerConfig.

PARAMETER	DESCRIPTION
name	Name of the circuit breaker instance. TYPE: str
config	The pre-built circuit breaker configuration. Use this path when the configuration is assembled at startup from a settings tree. The environment path is bypassed and the config is used as-is. TYPE: CircuitBreakerConfig

metrics

metrics() -> CircuitBreakerMetrics

Return current metrics for this circuit breaker.

restart async

restart() -> None

Restart the circuit breaker, clearing all counts and resetting to CLOSED state.

transition_to_closed async

transition_to_closed() -> None

Transition the circuit breaker to CLOSED state.

transition_to_open async

transition_to_open(until: float | None = None) -> None

Transition the circuit breaker to OPEN state.

transition_to_half_open async

transition_to_half_open() -> None

Transition the circuit breaker to HALF_OPEN state.

transition_to_forced_open async

transition_to_forced_open() -> None

Transition the circuit breaker to FORCED_OPEN state.

transition_to_forced_closed async

transition_to_forced_closed() -> None

Transition the circuit breaker to FORCED_CLOSED state.

CircuitBreakerError

CircuitBreakerError(
    *,
    name: str,
    last_error_time: datetime | None = None,
    last_error: Exception | None = None,
)

Bases: ResilienceError

Circuit breaker error.

Raised when calls are not permitted by the circuit breaker.

Initialize the error.

name instance-attribute

name = name

last_error instance-attribute

last_error = last_error

last_error_time instance-attribute

last_error_time = last_error_time

CircuitBreakerMetrics

Bases: BaseModel

Circuit breaker metrics.

name instance-attribute

name: str

state instance-attribute

state: CircuitBreakerState

active_calls instance-attribute

active_calls: int

total_error_count instance-attribute

total_error_count: int

total_success_count instance-attribute

total_success_count: int

consecutive_error_count instance-attribute

consecutive_error_count: int

consecutive_success_count instance-attribute

consecutive_success_count: int

last_error instance-attribute

last_error: ErrorDetails | None

CircuitBreakerState

Bases: StrEnum

Circuit breaker state.

State machine diagram:

┌────────┐ errors >= threshold  ┌────────┐
│ CLOSED │────────────────────> │  OPEN  │ <─┐
└────────┘                      └────────┘   │
    ▲                       timeout │        │ errors >= threshold
    │                               ▼        │
    │                         ┌───────────┐  │
    └─────────────────────────│ HALF_OPEN │──┘
      success >= threshold    └───────────┘

CLOSED class-attribute instance-attribute

CLOSED = 'CLOSED'

Circuit is closed, calls are allowed.

OPEN class-attribute instance-attribute

OPEN = 'OPEN'

Circuit is open, calls are not allowed.

HALF_OPEN class-attribute instance-attribute

HALF_OPEN = 'HALF_OPEN'

Circuit is half-open, calls are limited.

FORCED_OPEN class-attribute instance-attribute

FORCED_OPEN = 'FORCED_OPEN'

Circuit is open for an indefinite time, calls are not allowed.

FORCED_CLOSED class-attribute instance-attribute

FORCED_CLOSED = 'FORCED_CLOSED'

Circuit is forced closed for an indefinite time, calls are allowed.

ErrorDetails

Bases: BaseModel

Details about an error recorded by the circuit breaker.

time instance-attribute

time: datetime

type instance-attribute

type: str

msg instance-attribute

msg: str

GCRAConfig

Bases: _BaseRateLimiterConfig

Generic Cell Rate Algorithm: sliding-window rate limiting.

Stores a single timestamp per key (about 72 bytes). It is mathematically equivalent to the "leaky bucket" algorithm. If you are looking for a "leaky bucket" rate limiter, use GCRAConfig.

Use this when you need a precise sliding window, such as for HTTP API throttling with RFC 9211 RateLimit-* headers or legacy X-RateLimit-* headers. For the pattern "allow a burst of N, then 1 per second", use TokenBucketConfig instead.

Example:

from grelmicro.resilience import GCRAConfig, RateLimiter

# 5 requests per 60-second sliding window.
rl = RateLimiter("auth", GCRAConfig(limit=5, window=60))

Read more in the Rate Limiter docs.

type class-attribute instance-attribute

type: Literal['gcra'] = 'gcra'

Discriminator for the algorithm Pydantic union.

limit instance-attribute

limit: PositiveInt

Maximum number of requests allowed per window.

window instance-attribute

window: PositiveFloat

Window duration in seconds.

MemoryTokenBucket

MemoryTokenBucket(*, capacity: int, refill_rate: float)

Standalone in-memory token bucket.

Public, synchronous, thread-safe, and keyed. Use this class directly when you need fast, in-process, burst-friendly rate limiting in synchronous code. A typical use is inside a logging.Filter.

For async workflows, distributed coordination, or alternative algorithms, use RateLimiter with a backend instead.

Example:

from grelmicro.resilience.memory import MemoryTokenBucket

bucket = MemoryTokenBucket(capacity=5, refill_rate=1)


def handle(key: str) -> bool:
    return bucket.try_acquire(key=key)

Read more in the Rate Limiter docs.

Initialize the token bucket.

PARAMETER	DESCRIPTION
capacity	Maximum burst size. The bucket never holds more than capacity tokens. TYPE: int
refill_rate	Tokens replenished per second, up to capacity. TYPE: float

capacity property

capacity: int

Configured bucket capacity.

refill_rate property

refill_rate: float

Configured refill rate (tokens per second).

try_acquire

try_acquire(key: str = '', *, cost: float = 1.0) -> bool

Try to consume cost tokens for key.

Returns True and deducts the cost when the bucket has enough tokens, otherwise False (nothing deducted).

PARAMETER	DESCRIPTION
key	Identifier of the bucket (e.g. logger name, user id). TYPE: str DEFAULT: ''
cost	Tokens to consume. Must be > 0 and <= capacity. TYPE: float DEFAULT: 1.0

peek

peek(key: str = '') -> float

Return the current token count without consuming any.

PARAMETER	DESCRIPTION
key	Identifier of the bucket. TYPE: str DEFAULT: ''

reset

reset(key: str = '') -> None

Delete state for key, restoring full capacity.

PARAMETER	DESCRIPTION
key	Identifier of the bucket to reset. TYPE: str DEFAULT: ''

RateLimiter

RateLimiter(
    name: str,
    config: RateLimiterConfig,
    *,
    backend: RateLimiterBackend | str | None = None,
)

Rate limiter with a pluggable algorithm.

Most Python call sites should use the factory classmethods: RateLimiter.token_bucket for burst-friendly semantics or RateLimiter.gcra for precise sliding-window semantics.

Construct it directly with the instance name and a discriminated algorithm configuration when a config object already exists: TokenBucketConfig for burst-friendly semantics, or GCRAConfig for precise sliding-window semantics.

The algorithm is bound to the backend once at construction via RateLimiterBackend.bind. Each call to acquire, peek, or reset then runs the bound strategy directly. There is no extra algorithm lookup on each call.

Example:

from grelmicro.resilience import RateLimiter
from grelmicro.resilience.memory import MemoryRateLimiterBackend

MemoryRateLimiterBackend()
api = RateLimiter.token_bucket("api", capacity=10, refill_rate=1)


async def handle(user_id: str) -> None:
    result = await api.acquire(key=user_id)
    if not result.allowed:
        raise RuntimeError("rate limited")

Read more in the Rate Limiter docs.

Initialize the rate limiter.

PARAMETER	DESCRIPTION
name	The name of the rate limiter instance. Acts as the instance identity. Used as the key prefix on the backend and exposed via the name property. TYPE: str
config	The algorithm configuration. Most callers should prefer the RateLimiter.token_bucket or RateLimiter.gcra factory classmethods. Pass a config directly when it is already assembled elsewhere, for example from YAML or a pydantic-settings tree. Pass a TokenBucketConfig or a GCRAConfig. Both carry algorithm parameters plus the shared fail_open setting. The classes share a discriminated type field so serialization round-trips and pydantic-settings composition both work. TYPE: RateLimiterConfig
backend	An explicit backend instance. When None (the default), the registered backend is used. Set this to skip the global registry, for example in tests or when running several backends at the same time. TYPE: RateLimiterBackend | str | None DEFAULT: None

name property

name: str

Return the rate limiter identity.

config property

config: RateLimiterConfig

Return the algorithm configuration.

backend property

backend: RateLimiterBackend

Bound rate-limiter backend, resolved on each call.

When a backend instance was passed at construction it is always returned. Otherwise the registry is consulted on every access so that task-scoped resilience.use(...) overrides take effect.

from_config classmethod

from_config(
    name: str,
    config: RateLimiterConfig,
    *,
    backend: RateLimiterBackend | str | None = None,
) -> Self

Construct a RateLimiter from a name and a pre-built config.

Equivalent to RateLimiter(name, config, backend=backend). Use this when configuration is assembled declaratively at startup and the simple factory classmethods are not the right fit.

PARAMETER	DESCRIPTION
name	The name of the rate limiter instance. TYPE: str
config	The pre-built algorithm configuration. TYPE: RateLimiterConfig
backend	An explicit backend instance. When None (the default), the registered backend is used. TYPE: RateLimiterBackend | str | None DEFAULT: None

token_bucket classmethod

token_bucket(
    name: str,
    *,
    capacity: PositiveInt,
    refill_rate: PositiveFloat,
    fail_open: bool = False,
    backend: RateLimiterBackend | str | None = None,
) -> Self

Construct a token-bucket rate limiter.

Convenience factory for the common case. Builds a TokenBucketConfig internally and forwards to the constructor.

PARAMETER	DESCRIPTION
name	The name of the rate limiter instance. TYPE: str
capacity	Maximum burst size. The bucket holds at most capacity tokens. TYPE: PositiveInt
refill_rate	Tokens replenished per second, up to capacity. TYPE: PositiveFloat
fail_open	When True, the rate limiter returns an allowed result if the backend raises an error. TYPE: bool DEFAULT: False
backend	An explicit backend instance. When None (the default), the registered backend is used. TYPE: RateLimiterBackend | str | None DEFAULT: None

gcra classmethod

gcra(
    name: str,
    *,
    limit: PositiveInt,
    window: PositiveFloat,
    fail_open: bool = False,
    backend: RateLimiterBackend | str | None = None,
) -> Self

Construct a GCRA (sliding-window) rate limiter.

Convenience factory for the common case. Builds a GCRAConfig internally and forwards to the constructor.

PARAMETER	DESCRIPTION
name	The name of the rate limiter instance. TYPE: str
limit	Maximum number of requests allowed per window. TYPE: PositiveInt
window	Window duration in seconds. TYPE: PositiveFloat
fail_open	When True, the rate limiter returns an allowed result if the backend raises an error. TYPE: bool DEFAULT: False
backend	An explicit backend instance. When None (the default), the registered backend is used. TYPE: RateLimiterBackend | str | None DEFAULT: None

acquire async

acquire(*, key: str, cost: int = 1) -> RateLimitResult

Check rate limit and consume tokens if allowed.

PARAMETER	DESCRIPTION
key	Identifier for rate limiting (e.g. IP address, user ID, session). TYPE: str
cost	Number of tokens to consume. TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
RateLimitResult	RateLimitResult with allowed, limit, remaining,
RateLimitResult	retry_after, and reset_after fields.

RAISES	DESCRIPTION
ValueError	If cost is not between 1 and the algorithm's limit/capacity.

acquire_or_raise async

acquire_or_raise(
    *, key: str, cost: int = 1
) -> RateLimitResult

Check rate limit, raise if exceeded.

PARAMETER	DESCRIPTION
key	Identifier for rate limiting (e.g. IP address, user ID, session). TYPE: str
cost	Number of tokens to consume. TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
RateLimitResult	RateLimitResult if allowed.

RAISES	DESCRIPTION
RateLimitExceededError	If the rate limit is exceeded.

peek async

peek(*, key: str) -> RateLimitResult

Check rate limit state without consuming tokens.

PARAMETER	DESCRIPTION
key	Identifier for rate limiting (e.g. IP address, user ID, session). TYPE: str

RETURNS	DESCRIPTION
RateLimitResult	RateLimitResult reflecting the current state.
RateLimitResult	allowed indicates whether the next acquire would
RateLimitResult	succeed.

reset async

reset(*, key: str) -> None

Delete rate limit state for a key, restoring full quota.

Idempotent: resetting a nonexistent key is a no-op.

PARAMETER	DESCRIPTION
key	Identifier for rate limiting (e.g. IP address, user ID, session). TYPE: str

RateLimiterBackend

Bases: Protocol

Protocol for rate-limiter storage backends.

A backend holds the storage for every rate limiter that uses it. It turns an algorithm into a strategy through bind. The returned RateLimiterStrategy is what a RateLimiter calls on each acquire, peek, or reset. No extra algorithm lookup happens at call time.

bind

bind(config: RateLimiterConfig) -> RateLimiterStrategy

Build a strategy for the given algorithm config.

Called exactly once per RateLimiter when it is created. The returned strategy shares storage with the backend. Later requests call the strategy methods directly, with no extra algorithm lookup.

PARAMETER	DESCRIPTION
config	The algorithm configuration (TokenBucketConfig or GCRAConfig). TYPE: RateLimiterConfig

RETURNS	DESCRIPTION
RateLimiterStrategy	A strategy bound to config and this backend's storage.

RateLimiterStrategy

Bases: Protocol

A rate-limiter strategy for a specific algorithm and backend.

Returned by RateLimiterBackend.bind. The algorithm settings are already stored in the strategy, so the methods only need key and cost. No extra algorithm lookup happens at call time.

acquire async

acquire(*, key: str, cost: int) -> RateLimitResult

Try to acquire rate limit tokens.

PARAMETER	DESCRIPTION
key	The rate limit key (e.g. IP, user ID, session). TYPE: str
cost	Number of tokens to consume. TYPE: int

RETURNS	DESCRIPTION
RateLimitResult	RateLimitResult with allowed, limit, remaining,
RateLimitResult	retry_after, and reset_after fields.

peek async

peek(*, key: str) -> RateLimitResult

Check rate limit state without consuming tokens.

PARAMETER	DESCRIPTION
key	The rate limit key. TYPE: str

RETURNS	DESCRIPTION
RateLimitResult	RateLimitResult reflecting the current state.

reset async

reset(*, key: str) -> None

Delete rate limit state for a key, restoring full quota.

PARAMETER	DESCRIPTION
key	The rate limit key to reset. TYPE: str

RateLimitExceededError

RateLimitExceededError(*, key: str, retry_after: float)

Bases: ResilienceError

Rate limit exceeded error.

Raised when a rate limit check fails (too many requests).

Initialize the error.

key instance-attribute

key = key

retry_after instance-attribute

retry_after = retry_after

RateLimitResult

Bases: NamedTuple

Result of a rate limit check.

Fields map to HTTP rate limit headers: - allowed -> 200 vs 429 status - limit -> X-RateLimit-Limit / RateLimit-Policy: ;q= - remaining -> X-RateLimit-Remaining / RateLimit: ;r= - retry_after -> Retry-After header - reset_after -> X-RateLimit-Reset / RateLimit: ;t=

allowed instance-attribute

allowed: bool

Whether the request is permitted.

limit instance-attribute

limit: int

Total quota (capacity for TokenBucketConfig, limit for GCRAConfig).

remaining instance-attribute

remaining: int

Remaining tokens or requests.

retry_after instance-attribute

retry_after: float

Seconds until the next request is allowed (0.0 if allowed).

reset_after instance-attribute

reset_after: float

Seconds until the full quota resets.

ResilienceError

Bases: GrelmicroError

Base class for all resilience-related errors.

This class serves as the base for all errors related to resilience mechanisms such as circuit breakers, retries, etc.

ResilienceSettingsValidationError

ResilienceSettingsValidationError(
    error: ValidationError | str,
)

Bases: ResilienceError, SettingsValidationError

Resilience Settings Validation Error.

TokenBucketConfig

Bases: _BaseRateLimiterConfig

Classic token bucket rate-limiting algorithm.

The bucket starts full and refills continuously at refill_rate tokens per second, capped at capacity. Each request consumes tokens. If the bucket has enough, the request is allowed, otherwise it is rejected with a retry_after hint.

Use this when you want the pattern "allow a burst of N requests, then a steady rate of 1 request per second". The token bucket is a common choice for bursty rate limiting.

Example:

from grelmicro.resilience import RateLimiter, TokenBucketConfig

# Allow 10 in a burst, then 1/sec sustained.
rl = RateLimiter("api", TokenBucketConfig(capacity=10, refill_rate=1))

Read more in the Rate Limiter docs.

type class-attribute instance-attribute

type: Literal['token_bucket'] = 'token_bucket'

Discriminator for the algorithm Pydantic union.

capacity instance-attribute

capacity: PositiveInt

Maximum burst size. The bucket never holds more than capacity tokens.

refill_rate instance-attribute

refill_rate: PositiveFloat

Tokens replenished per second, up to capacity.

grelmicro.resilience.memory

Memory Rate Limiter Backend and standalone primitives.

MemoryRateLimiterBackend

MemoryRateLimiterBackend()

Bases: RateLimiterBackend

In-memory rate limiter backend.

Supports both TokenBucketConfig and GCRAConfig algorithm configs. State is held in separate per-algorithm maps so two rate limiters with the same name but different algorithms cannot collide. Thread-safe.

Use it for tests and single-process deployments. For distributed coordination, use RedisRateLimiterBackend.

Example:

from grelmicro.resilience import RateLimiter, TokenBucketConfig, use_backend
from grelmicro.resilience.memory import MemoryRateLimiterBackend

use_backend(MemoryRateLimiterBackend())
rl = RateLimiter("api", TokenBucketConfig(capacity=10, refill_rate=1))

Read more in the Rate Limiter docs.

Initialize the rate limiter backend.

bind

bind(config: RateLimiterConfig) -> RateLimiterStrategy

Build a strategy for the given algorithm config.

Called once by RateLimiter when the rate limiter is created. This is the only place that picks which algorithm to run. Later calls to acquire, peek, and reset use the returned strategy directly.

grelmicro.resilience.redis

Redis Rate Limiter Backend.

RedisRateLimiterBackend

RedisRateLimiterBackend(
    url: RedisDsn | str | None = None, *, prefix: str = ""
)

Bases: RateLimiterBackend

Redis rate limiter backend.

Supports both TokenBucketConfig and GCRAConfig algorithm configs via atomic Lua scripts. Safe across processes and machines.

Example:

from grelmicro.resilience import RateLimiter, TokenBucketConfig
from grelmicro.resilience.redis import RedisRateLimiterBackend


async def main() -> None:
    async with RedisRateLimiterBackend("redis://localhost:6379/0"):
        rl = RateLimiter(
            "api",
            TokenBucketConfig(capacity=10, refill_rate=1),
        )
        await rl.acquire(key="u1")

Read more in the Rate Limiter docs.

Initialize the rate limiter backend.

PARAMETER	DESCRIPTION
url	The Redis URL. If not provided, the URL will be taken from the environment variables REDIS_URL or REDIS_HOST, REDIS_PORT, REDIS_DB, and REDIS_PASSWORD. TYPE: RedisDsn | str | None DEFAULT: None
prefix	Prefix prepended to every Redis key the backend writes. Use it to avoid collisions with other consumers of the same Redis database. By default no prefix is added. TYPE: str DEFAULT: ''

bind

bind(config: RateLimiterConfig) -> RateLimiterStrategy

Build a strategy for the given algorithm config.

Each strategy has its own Lua scripts. It registers them with the Redis client when the strategy is created.