Resilience
- Start here: Resilience guide
- Common recipes: Rate Limiter, Circuit Breaker, Retry, Fallback, Timeout, Shield
- Configuration: Composing patterns
Resilience.
Pick the front door first, the algorithm second, the backend third.
Front doors (start here):
RateLimiter.token_bucket(...)orRateLimiter.sliding_window(...)for rate limiting.CircuitBreaker("name")orCircuitBreaker.consecutive_count(...)for circuit breaking.Retry("name", backoff, when=...)or the@retry(...)/retrying(...)decorator and block form.Fallback("name", when=..., default=...)or the@fallback(...)/falling_back(...)decorator and block form.Timeout("name", seconds=...)for deadlines.Bulkhead("name", max_concurrent=...)or the@bulkheaddecorator to cap concurrent in-flight calls.Shield("name")or the@shield(...)decorator for the bundled timeout + retry + adaptive rate-limit + cache + fallback profile.
Components (wire the front doors into Grelmicro(uses=[...])):
RateLimiterRegistry(backend)andCircuitBreakerRegistry(backend). They register the shared storage. Without them, passbackend=on the primitive (a memory adapter is fine for tests and single-replica services).
Adapters / backends (one per storage choice, used inside
RateLimiterRegistry / CircuitBreakerRegistry): MemoryRateLimiterAdapter,
RedisRateLimiterAdapter, PostgresRateLimiterAdapter,
SQLiteRateLimiterAdapter, MemoryCircuitBreakerAdapter,
RedisCircuitBreakerAdapter, PostgresCircuitBreakerAdapter. End
users rarely name these directly. The Components do.
Configs (frozen Pydantic models, accept env vars):
TokenBucketConfig, SlidingWindowConfig,
CircuitBreakerConfig, RetryConfig, FallbackConfig,
TimeoutConfig, BulkheadConfig, ShieldConfig. One per pattern,
plus backoff configs (ExponentialBackoff, LinearBackoff, ...).
Loading: top-level re-exports are PEP 562 lazy. Importing this
package loads _components, _match, _outcome, _protocol,
and errors plus the eager exports listed below. Every other
pattern, its algorithm configs, and the memory/redis adapters
load on first attribute access. from grelmicro.resilience import
CircuitBreaker does not import anything related to RateLimiter.
Eager exports (loaded at package import because the function name
shadows a submodule of the same name): retry, retrying,
fallback, falling_back, shield. The shield import pulls the
full grelmicro.resilience.shield subpackage.
module-attribute
CircuitBreakerConfig = ConsecutiveCountConfig
Discriminated union of supported circuit-breaker algorithm configurations.
Single-arm today. Future algorithms (failure-rate, slow-call) join
the union via the kind discriminator without breaking existing
serialized configs.
module-attribute
Matcher = Callable[[Outcome[Any]], bool]
Callable signature every Match resolves to.
Returns True when the outcome should engage the strategy.
module-attribute
RateLimiterConfig = TokenBucketConfig | SlidingWindowConfig
Discriminated union of supported rate-limiter algorithm configurations.
module-attribute
RetryBackoffConfig = (
ExponentialBackoff
| ConstantBackoff
| LinearBackoff
| FibonacciBackoff
| RandomBackoff
)
Discriminated union of supported retry backoff configurations.
module-attribute
ShieldConfig = (
InternalShieldConfig
| ApiShieldConfig
| SlowShieldConfig
)
Discriminated union over the three Shield profile configs.
module-attribute
retrying = _RetryingFactory()
Bulkhead(
name: str,
*,
max_concurrent: PositiveInt | None = None,
max_wait: NonNegativeFloat | None = None,
max_workers: PositiveInt | None = None,
uses: Iterable[
AbstractAsyncContextManager[object]
] = (),
config: BulkheadConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[BulkheadConfig]
Bulkhead policy.
A named, reusable concurrency limiter with three-paths
configuration and live reconfiguration. Use it as an async context
manager or as a decorator on async functions to bound the number of
in-flight calls, and to_thread to run blocking work on a bounded
private thread pool.
When the bulkhead is full, a caller waits up to max_wait seconds
for a permit, then is rejected with
BulkheadFullError. The
default fails fast (no wait).
Read more in the Bulkhead docs.
Initialize the bulkhead.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the bulkhead. Used as the env namespace, the rejection error label, and the thread-name prefix.
TYPE:
|
max_concurrent
|
Maximum concurrent calls.
TYPE:
|
max_wait
|
Seconds to wait for a permit before rejecting.
TYPE:
|
max_workers
|
Private thread-pool size for
TYPE:
|
uses
|
Providers and Components, in the same shape as
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the bulkhead identity.
classmethod
from_config(name: str, config: BulkheadConfig) -> Self
Construct a Bulkhead from a name and a pre-built BulkheadConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the bulkhead.
TYPE:
|
config
|
The pre-built bulkhead configuration.
TYPE:
|
async
to_thread(
func: Callable[..., Any], /, *args: Any, **kwargs: Any
) -> Any
Run func in a worker thread, bounded by max_workers.
Routes through the bulkhead's private ThreadPoolExecutor when
max_workers is set, otherwise the event loop's shared executor
(asyncio.to_thread).
| PARAMETER | DESCRIPTION |
|---|---|
func
|
Blocking callable to run off the event loop.
TYPE:
|
Bases: BaseModel
Bulkhead policy configuration.
Frozen Pydantic data class. Three-paths configuration: kwargs, instance, or env vars.
Read more in the Bulkhead docs.
class-attribute
instance-attribute
max_concurrent: PositiveInt | None = None
Maximum concurrent calls admitted to the bulkhead. None (the default) leaves concurrency unbounded.
class-attribute
instance-attribute
max_wait: NonNegativeFloat | None = None
Seconds a caller waits for a free permit before the bulkhead rejects it with BulkheadFullError. None (the default) and 0 reject immediately (fail fast). Ignored when max_concurrent is None.
class-attribute
instance-attribute
max_workers: PositiveInt | None = None
Size of the private thread pool backing to_thread. None (the default) uses the event loop's shared executor.
BulkheadFullError(*, name: str, max_concurrent: int)
Bases: ResilienceError, AdmissionError
Bulkhead full error.
Raised when a bulkhead has no free permit and the caller's
max_wait elapsed (or was zero, the fail-fast default).
Initialize the error.
instance-attribute
name = name
instance-attribute
max_concurrent = max_concurrent
CircuitBreakerRegistry(
source: Provider
| CircuitBreakerBackend
| type[Provider | CircuitBreakerBackend],
*,
name: str = "default",
)
CircuitBreakerBackend wrapper exposing (circuitbreaker, name) registration.
Registered on a Grelmicro app via Grelmicro(uses=[CircuitBreakerRegistry(redis)]).
The active app resolves CircuitBreaker patterns to this Component's
backend on every call.
Accepts a Provider or a CircuitBreakerBackend. When given a Provider,
the component calls provider.circuitbreaker() to build the matching adapter.
Example
from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import CircuitBreaker, CircuitBreakerRegistry
redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis, CircuitBreakerRegistry(redis)])
payment = CircuitBreaker("payment")
async with micro:
async with payment:
...
Initialize the Component with the wrapped backend.
| PARAMETER | DESCRIPTION |
|---|---|
source
|
A
TYPE:
|
name
|
Registration name. Multiple
TYPE:
|
class-attribute
kind: str = 'circuitbreaker'
property
name: str
Return the registration name.
CircuitBreaker(
name: str,
config: CircuitBreakerConfig | None = None,
*,
backend: CircuitBreakerBackend | str | None = None,
)
Bases: Reconfigurable['CircuitBreakerConfig']
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.
Supports live reconfiguration via
reconfigure(new_config).
A swap takes effect on the next call. In-flight calls keep the
config they entered with. The current state, counters, and
last_error are kept. A new log_level is applied to the
logger. See Live reconfiguration.
Initialize the circuit breaker, defaulting the algorithm to consecutive-count.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the circuit breaker instance. Acts as the instance identity, exposed via the
TYPE:
|
config
|
The algorithm configuration. Defaults to Today the discriminated union has a single arm:
TYPE:
|
backend
|
The circuit breaker backend that owns the lifespan and (with a shared adapter) the cross-replica state. Accepts a backend instance, the name of a registered
backend (e.g.
TYPE:
|
property
backend: CircuitBreakerBackend
Bound circuit breaker backend, resolved on each call.
Resolution order:
1. An explicit backend= passed at construction wins.
2. The active Grelmicro app is consulted via
Grelmicro.current() so that micro.override(...) blocks
take effect.
| RAISES | DESCRIPTION |
|---|---|
OutOfContextError
|
No backend resolved in this scope. Pass
|
property
from_thread: _ThreadAdapter
Sync adapter for use from a worker thread.
Use it from a synchronous handler that the host framework runs in a worker thread. The adapter signals the intent explicitly so the async API stays the documented default.
property
name: str
Return the name of the circuit breaker.
property
last_error: Exception | None
Return the last error recorded by the circuit breaker.
property
last_error_time: datetime | None
Return the time of the last error recorded by the circuit breaker.
classmethod
from_config(
name: str,
config: CircuitBreakerConfig,
*,
backend: CircuitBreakerBackend | str | None = None,
) -> Self
Construct a CircuitBreaker from a name and a pre-built CircuitBreakerConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the circuit breaker instance.
TYPE:
|
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:
|
backend
|
The circuit breaker backend.
TYPE:
|
classmethod
consecutive_count(
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,
backend: CircuitBreakerBackend | str | None = None,
) -> Self
Construct a CircuitBreaker running the consecutive-count algorithm.
Sibling of from_config
and the bare constructor: the bare constructor reads env vars
for unset fields, this factory does not.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the circuit breaker instance.
TYPE:
|
ignore_exceptions
|
Exceptions ignored by the breaker.
TYPE:
|
error_threshold
|
Consecutive errors before the breaker opens. Default: 5.
TYPE:
|
success_threshold
|
Consecutive successes in
TYPE:
|
reset_timeout
|
Seconds the breaker stays
TYPE:
|
half_open_capacity
|
Maximum concurrent calls allowed in the
TYPE:
|
log_level
|
Logging level for state-change messages. Default:
TYPE:
|
backend
|
The circuit breaker backend.
TYPE:
|
async
isolate() -> None
Force the breaker open and keep it open until reset.
The manual "big red button". Moves the breaker to
FORCED_OPEN, so every call is blocked with
CircuitBreakerError regardless of outcomes, until
reset returns it
to automatic operation.
async
reset() -> None
Return the breaker to normal automatic operation.
Clears all counters and the last recorded error, then moves the
breaker to CLOSED. Use it to release an
isolate hold or
to start fresh from a known state.
Bases: Protocol
Protocol for circuit-breaker storage backends.
A backend owns the lifespan boundary and the storage for every
circuit breaker bound to it. It turns a name and a config into a
strategy through
bind. The
returned
CircuitBreakerStrategy
is what a
CircuitBreaker calls on
each try_acquire, record_outcome, transition, or
get_snapshot. No extra algorithm lookup happens at call time.
Implementations capture the running event loop on __aenter__
in a _loop attribute so the sync from_thread adapter can
dispatch coroutines back into it.
class-attribute
is_shared: bool
Whether the backend stores state outside the local process.
True for distributed backends (e.g. Redis), False for
process-local backends (e.g. memory). User code can read this
to decide whether last_error is per-replica or fleet-wide.
bind(
*, name: str, config: CircuitBreakerConfig
) -> CircuitBreakerStrategy
Build a strategy for the named breaker and algorithm config.
Called once per
CircuitBreaker when
it is created, and again whenever the breaker's config changes
through live reconfiguration.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Breaker name, used as the storage key in shared backends.
TYPE:
|
config
|
Algorithm configuration for the breaker.
TYPE:
|
CircuitBreakerError(
*,
name: str,
last_error_time: datetime | None = None,
last_error: Exception | None = None,
)
Bases: ResilienceError, AdmissionError
Circuit breaker error.
Raised when calls are not permitted by the circuit breaker.
Initialize the error.
instance-attribute
name = name
instance-attribute
last_error = last_error
instance-attribute
last_error_time = last_error_time
dataclass
CircuitBreakerMetrics(
name: str,
state: CircuitBreakerState,
active_calls: int,
total_error_count: int,
total_success_count: int,
consecutive_error_count: int,
consecutive_success_count: int,
last_error: ErrorDetails | None,
)
Circuit breaker metrics.
instance-attribute
name: str
instance-attribute
active_calls: int
instance-attribute
total_error_count: int
instance-attribute
total_success_count: int
instance-attribute
consecutive_error_count: int
instance-attribute
consecutive_success_count: int
Bases: NamedTuple
Snapshot of circuit breaker state returned by a strategy.
Returned by every
CircuitBreakerStrategy
method that mutates or reads state. The breaker uses it to refresh
its local cache so reads of cb.state and cb.metrics() reflect
the latest truth from the strategy.
Algorithm-specific counters (consecutive_error_count,
consecutive_success_count) are populated by the consecutive-count
algorithm. Future algorithms may populate additional fields.
instance-attribute
opened_at: float
Strategy-clock seconds when the breaker entered OPEN. 0.0 when not OPEN.
Each strategy picks its own clock. Treat this as a relative value: only compare with timestamps emitted by the same strategy.
class-attribute
instance-attribute
consecutive_error_count: int = 0
Consecutive errors recorded by the strategy. Consecutive-count algorithm only.
class-attribute
instance-attribute
consecutive_success_count: int = 0
Consecutive successes recorded by the strategy. Consecutive-count algorithm only.
Bases: StrEnum
Circuit breaker state.
State machine diagram:
┌────────┐ errors >= threshold ┌────────┐
│ CLOSED │────────────────────> │ OPEN │ <─┐
└────────┘ └────────┘ │
▲ timeout │ │ errors >= threshold
│ ▼ │
│ ┌───────────┐ │
└─────────────────────────│ HALF_OPEN │──┘
success >= threshold └───────────┘
class-attribute
instance-attribute
CLOSED = 'CLOSED'
Circuit is closed, calls are allowed.
class-attribute
instance-attribute
OPEN = 'OPEN'
Circuit is open, calls are not allowed.
class-attribute
instance-attribute
HALF_OPEN = 'HALF_OPEN'
Circuit is half-open, calls are limited.
class-attribute
instance-attribute
FORCED_OPEN = 'FORCED_OPEN'
Circuit is open for an indefinite time, calls are not allowed.
class-attribute
instance-attribute
FORCED_CLOSED = 'FORCED_CLOSED'
Circuit is forced closed for an indefinite time, calls are allowed.
Bases: Protocol
A circuit-breaker strategy for a specific algorithm and backend.
Returned by
CircuitBreakerBackend.bind.
The breaker name and algorithm settings are already stored in the
strategy, so the methods take no extra arguments beyond the call's
own data (outcome, manual transition target).
async
try_acquire() -> bool
Attempt to admit a call.
Returns True when the call is admitted. Returns False when the breaker is OPEN or FORCED_OPEN, or when HALF_OPEN has no remaining capacity.
async
record_outcome(
*, success: bool, duration: float = 0.0
) -> CircuitBreakerSnapshot
Record a call outcome and return the resulting snapshot.
| PARAMETER | DESCRIPTION |
|---|---|
success
|
Whether the call completed without an error that counts against the breaker.
TYPE:
|
duration
|
Wall-clock seconds the call took. Consumed by algorithms that classify slow calls. Ignored by the consecutive-count algorithm.
TYPE:
|
async
transition(
*,
desired: CircuitBreakerState,
cool_down: float | None = None,
) -> None
Force the breaker into desired.
| PARAMETER | DESCRIPTION |
|---|---|
desired
|
Target state to force the breaker into.
TYPE:
|
cool_down
|
Seconds the breaker stays OPEN before moving to HALF_OPEN.
TYPE:
|
async
get_snapshot() -> CircuitBreakerSnapshot
Return the current snapshot without mutating state.
Bases: BaseModel
Consecutive-count circuit breaker algorithm.
Opens after error_threshold consecutive failures. Closes from
HALF_OPEN after success_threshold consecutive successes. A
single success in CLOSED resets the running error count.
Use this when failures cluster, for example transient downstream
outages where the first N errors in a row are a strong signal. For
failure-rate or slow-call detection, plug in a future algorithm
config through the same kind discriminator.
Example:
from grelmicro.resilience import CircuitBreaker, ConsecutiveCountConfig
cb = CircuitBreaker.from_config(
"payments",
ConsecutiveCountConfig(error_threshold=5, reset_timeout=30.0),
)
Read more in the Circuit Breaker docs.
class-attribute
instance-attribute
kind: Literal['consecutive_count'] = 'consecutive_count'
Discriminator for the algorithm Pydantic union.
class-attribute
instance-attribute
ignore_exceptions: tuple[
ImportString[type[Exception]], ...
] = ()
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" for YAML and env loading.
Env vars accept comma-separated values or JSON arrays.
class-attribute
instance-attribute
error_threshold: PositiveInt = 5
Consecutive errors before the breaker opens.
class-attribute
instance-attribute
success_threshold: PositiveInt = 2
Consecutive successes in HALF_OPEN state before the breaker closes.
class-attribute
instance-attribute
reset_timeout: PositiveFloat = 30.0
Seconds the breaker stays OPEN before transitioning to HALF_OPEN.
class-attribute
instance-attribute
half_open_capacity: PositiveInt = 1
Maximum concurrent calls allowed in the HALF_OPEN state.
class-attribute
instance-attribute
log_level: LogLevel = 'WARNING'
Logging level for state-change messages.
Bases: BaseModel
Constant delay between retries.
Use this for polling-style retries where you wait a fixed
interval. For network and HTTP calls, prefer
ExponentialBackoff
to avoid synchronized retry storms.
Example:
from grelmicro.resilience import ConstantBackoff, Retry
policy = Retry(
"wait_job",
ConstantBackoff(delay=1.0),
when=NotReady,
attempts=20,
)
Read more in the Retry docs.
class-attribute
instance-attribute
kind: Literal['constant'] = 'constant'
Discriminator for the backoff Pydantic union.
class-attribute
instance-attribute
delay: PositiveFloat = 1.0
Fixed delay in seconds between retries.
dataclass
ErrorDetails(time: datetime, type: str, msg: str)
Details about an error recorded by the circuit breaker.
instance-attribute
time: datetime
instance-attribute
type: str
instance-attribute
msg: str
Bases: BaseModel
Exponential backoff with optional jitter.
The raw delay before retry N is
min(base_delay * 2 ** (N - 1), max_delay): it doubles each
attempt until it reaches the cap. jitter then transforms
that raw delay so concurrent callers do not retry in lockstep
(the actual sleep may be smaller than the raw value).
Example:
from grelmicro.resilience import ExponentialBackoff, Retry
policy = Retry(
"payments",
ExponentialBackoff(base_delay=0.2, max_delay=10.0),
when=httpx.HTTPError,
)
Read more in the Retry docs.
class-attribute
instance-attribute
kind: Literal['exponential'] = 'exponential'
Discriminator for the backoff Pydantic union.
class-attribute
instance-attribute
base_delay: PositiveFloat = 0.1
Initial delay in seconds before the first retry.
class-attribute
instance-attribute
max_delay: PositiveFloat = 30.0
Maximum delay in seconds. Caps the exponential growth.
class-attribute
instance-attribute
jitter: Literal["none", "full", "equal", "decorrelated"] = (
"full"
)
Jitter mode. full (default) samples from [0, raw] and is the safest choice against retry storms. equal samples from [raw/2, raw] and keeps timing more predictable. decorrelated chains samples across attempts and is best for many clients hitting the same recovering dependency. none disables jitter.
Fallback(
name: str,
*,
when: WhenInput | None = None,
default: Any = _UNSET,
factory: Callable[[BaseException], Any] | None = None,
config: FallbackConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[FallbackConfig]
Fallback policy.
A named, reusable fallback policy with three-paths configuration
and live reconfiguration. Use the constructor for the common
case and Fallback.from_config when configuration is assembled
elsewhere.
Read more in the Fallback docs.
Initialize the fallback policy.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the fallback policy. Used as the env namespace and exposed via the
TYPE:
|
when
|
Exception filter that engages the fallback. Pass a
TYPE:
|
default
|
Static fallback value. Mutually exclusive with
TYPE:
|
factory
|
Callable that produces the fallback value from the matched exception. Mutually exclusive with
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the fallback policy identity.
classmethod
from_config(name: str, config: FallbackConfig) -> Self
Construct a Fallback from a name and a pre-built FallbackConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the fallback policy.
TYPE:
|
config
|
The pre-built fallback configuration.
TYPE:
|
Bases: BaseModel
Fallback policy configuration.
Holds the exception filter plus exactly one of default or
factory. Frozen Pydantic data class. Three-paths configuration:
kwargs, instance, or env vars.
Read more in the Fallback docs.
instance-attribute
when: Match
Exception filter that engages the fallback. Pass a Match or a shorthand: an exception class, a tuple of classes, or a predicate on the exception. BaseException subclasses outside Exception are never caught.
class-attribute
instance-attribute
default: Any = _UNSET
Static value returned when when matches. Mutually exclusive with factory. Exactly one must be set.
class-attribute
instance-attribute
factory: Callable[[BaseException], Any] | None = None
Callable that produces the fallback value from the exception. Mutually exclusive with default.
class-attribute
instance-attribute
model_config = {'arbitrary_types_allowed': True}
FallbackResult()
Holder for the value produced inside a falling_back block.
Call set(value)
on success. When the block raises an exception matching when=,
the exception is suppressed and the holder is filled with the
configured default (or the factory output). Access the resulting
value with the
value property
after the block exits.
Initialize an empty result holder.
property
value: T
Return the recorded value (success or fallback).
| RAISES | DESCRIPTION |
|---|---|
RuntimeError
|
When accessed before any value was set. |
set(value: T) -> None
Record the success value for the block.
Bases: BaseModel
Fibonacci backoff: delays follow the Fibonacci sequence scaled by base_delay.
The delay before retry N is min(base_delay * fib(N), max_delay),
where fib(1) = 1, fib(2) = 1, fib(3) = 2, fib(4) = 3, ...
Sits between linear and exponential. Slower than exponential but eventually outpaces linear. Used historically in TCP congestion control and in retry strategies generally.
For most retries, prefer
ExponentialBackoff.
Reach for Fibonacci when exponential's growth is too aggressive
and linear's is too slow.
Example:
from grelmicro.resilience import FibonacciBackoff, Retry
# 1s, 1s, 2s, 3s, 5s, 8s, ...
policy = Retry(
"deferred",
FibonacciBackoff(base_delay=1.0, max_delay=60.0),
when=ServiceError,
attempts=8,
)
Read more in the Retry docs.
class-attribute
instance-attribute
kind: Literal['fibonacci'] = 'fibonacci'
Discriminator for the backoff Pydantic union.
class-attribute
instance-attribute
base_delay: PositiveFloat = 1.0
Multiplier in seconds applied to each Fibonacci term.
class-attribute
instance-attribute
max_delay: PositiveFloat = 30.0
Maximum delay in seconds. Caps the Fibonacci growth.
Bases: BaseModel
Linear backoff: delay grows by base_delay each attempt.
The delay before retry N is min(base_delay * N, max_delay).
Use this for predictable progression when you have a rough idea
of the recovery time and want a smooth ramp without exponential
blow-up. Common for polling that escalates over time.
For network and HTTP calls, prefer
ExponentialBackoff
to avoid synchronized retry storms.
Example:
from grelmicro.resilience import LinearBackoff, Retry
# 1s, 2s, 3s, 4s, ...
policy = Retry(
"ramp",
LinearBackoff(base_delay=1.0, max_delay=10.0),
when=ServiceError,
attempts=5,
)
Read more in the Retry docs.
class-attribute
instance-attribute
kind: Literal['linear'] = 'linear'
Discriminator for the backoff Pydantic union.
class-attribute
instance-attribute
base_delay: PositiveFloat = 1.0
Increment in seconds added per attempt.
class-attribute
instance-attribute
max_delay: PositiveFloat = 30.0
Maximum delay in seconds. Caps the linear growth.
Match(matcher: Matcher, repr_: str)
Outcome filter that resilience strategies consume.
Build instances through the classmethods, never the constructor.
Compose with the | and & operators. Each primitive
matcher has a symmetric not_* twin for the negated form.
Read more in the Retry filtering docs.
explain() -> str
Return the human-readable matcher tree for debugging.
classmethod
exception(
*exception_types_or_predicate: type[Exception]
| Callable[[Exception], bool],
) -> Match
Engage when the call raised a matching exception.
Pass one or more exception classes, or a single callable
predicate (Exception) -> bool. When mixed forms are
passed (some classes, some callables), the result raises
TypeError at construction.
classmethod
result(
value_or_predicate: Any | Callable[[Any], bool],
) -> Match
Engage when the call returned a matching value.
Pass a literal value (compared with ==) or a callable
predicate (result) -> bool. Functions are always treated
as predicates: to match a function literal, wrap it in a
predicate (lambda r: r is my_fn).
classmethod
exception_message(
*,
contains: str | None = None,
regex: str | Pattern[str] | None = None,
) -> Match
Engage when the exception's message matches the predicate.
Pass exactly one of contains= (substring) or regex=
(compiled or string regex).
classmethod
exception_cause(
*exception_types_or_predicate: type[Exception]
| Callable[[BaseException | None], bool],
) -> Match
Engage when the exception's __cause__ matches.
Same shorthand as Match.exception: one or more classes,
or a single callable predicate on exc.__cause__.
classmethod
always() -> Match
Engage on every outcome.
Useful as the explicit "always retry" policy.
Note: BaseException subclasses outside Exception are
still never retried by the strategy itself, regardless of
the matcher.
classmethod
Engage when the predicate returns true for the outcome.
Use this when the filter must observe both the exception and
the result together. Most call sites should reach for
Match.exception or Match.result instead.
classmethod
not_exception(
*exception_types_or_predicate: type[Exception]
| Callable[[Exception], bool],
) -> Match
Engage when the call raised an exception that does NOT match.
Scoped to raised outcomes: returns False for any returned
outcome. Same arguments as Match.exception.
classmethod
not_result(
value_or_predicate: Any | Callable[[Any], bool],
) -> Match
Engage when the call returned a value that does NOT match.
Scoped to returned outcomes: returns False for any raised
outcome. Same argument as Match.result.
classmethod
not_exception_message(
*,
contains: str | None = None,
regex: str | Pattern[str] | None = None,
) -> Match
Engage when the exception's message does NOT match.
Scoped to raised outcomes: returns False for any returned
outcome.
classmethod
not_exception_cause(
*exception_types_or_predicate: type[Exception]
| Callable[[BaseException | None], bool],
) -> Match
Engage when the exception's __cause__ does NOT match.
Scoped to raised outcomes: returns False for any returned
outcome.
MemoryCircuitBreakerAdapter()
Bases: CircuitBreakerBackend
In-memory circuit breaker adapter.
State for every breaker bound to this adapter is held in process, keyed by breaker name. Closing the adapter clears every breaker's state so the next start begins on a clean slate.
Use it for tests and single-process deployments. Use
RedisCircuitBreakerAdapter for fleet-wide shared state.
Initialize the circuit breaker adapter.
class-attribute
is_shared: bool = False
bind(
*, name: str, config: ConsecutiveCountConfig
) -> CircuitBreakerStrategy
Build a strategy bound to this adapter's per-name state.
Two breakers constructed with the same name against the same
adapter share the same _BreakerState entry, mirroring the
Redis adapter's per-name keying.
MemoryRateLimiterAdapter()
Bases: RateLimiterBackend
In-memory rate limiter adapter.
Supports both
TokenBucketConfig
and SlidingWindowConfig
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
RedisRateLimiterAdapter.
Example:
from grelmicro.resilience import RateLimiterRegistry, RateLimiter
from grelmicro.resilience.ratelimiter.memory import MemoryRateLimiterAdapter
micro = Grelmicro(uses=[RateLimiterRegistry(MemoryRateLimiterAdapter())])
rl = RateLimiter.token_bucket("api", capacity=10, refill_rate=1)
Read more in the Rate Limiter docs.
Initialize the rate limiter adapter.
bind(
config: TokenBucketConfig | SlidingWindowConfig,
) -> 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.
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.ratelimiter.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
TYPE:
|
refill_rate
|
Tokens replenished per second, up to
TYPE:
|
property
capacity: int
Configured bucket capacity.
property
refill_rate: float
Configured refill rate (tokens per second).
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:
|
cost
|
Tokens to consume. Must be > 0 and <=
TYPE:
|
peek(key: str = '') -> float
Return the current token count without consuming any.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier of the bucket.
TYPE:
|
reset(key: str = '') -> None
Delete state for key, restoring full capacity.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier of the bucket to reset.
TYPE:
|
dataclass
Outcome(
exception: Exception | None,
result: T | None,
raised: bool,
)
Result of a protected call: an exception or a return value.
Exactly one of exception and result is meaningful per
instance, signalled by raised. The T type parameter is
the function's return type.
Read more in the Retry filtering docs.
instance-attribute
exception: Exception | None
The exception raised by the call, or None when it returned.
instance-attribute
result: T | None
The value returned by the call, or None when it raised.
instance-attribute
raised: bool
True when the call raised, False when it returned.
classmethod
from_exception(exception: Exception) -> Outcome[T]
Build an Outcome for a raised call.
PostgresCircuitBreakerAdapter(
*,
provider: PostgresProvider | None = None,
env_prefix: str = "POSTGRES_",
prefix: str = "",
table_name: str = "grelmicro_circuit_breaker",
auto_migrate: bool = True,
)
Bases: CircuitBreakerBackend
Postgres circuit breaker adapter.
Builds a per-breaker
CircuitBreakerStrategy
that stores state in a row of {table_name} keyed by breaker name.
Every admission and counter update runs inside a PL/pgSQL function
that holds pg_advisory_xact_lock for the breaker, so concurrent
replicas converge to the same state without coordination locks.
Today the consecutive-count algorithm is the only strategy. Future
algorithm configs plug in through the same bind entry point.
last_error and last_error_time stay per-replica.
Example:
from grelmicro import Grelmicro
from grelmicro.providers.postgres import PostgresProvider
from grelmicro.resilience import CircuitBreakerRegistry, CircuitBreaker
from grelmicro.resilience.circuitbreaker.postgres import (
PostgresCircuitBreakerAdapter,
)
postgres = PostgresProvider("postgresql://localhost:5432/app")
micro = Grelmicro(
uses=[
postgres,
CircuitBreakerRegistry(PostgresCircuitBreakerAdapter(provider=postgres)),
]
)
payments = CircuitBreaker("payments")
Read more in the Circuit Breaker docs.
Initialize the circuit breaker adapter.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
A pre-built
TYPE:
|
env_prefix
|
Environment variable prefix used by the implicit
TYPE:
|
prefix
|
Prefix prepended to every breaker name the adapter writes. Use it to avoid collisions with other consumers of the same Postgres table.
TYPE:
|
table_name
|
Table that stores circuit-breaker state. Auto-created
on first connect (set
TYPE:
|
auto_migrate
|
When True (the default), the adapter creates the table
and SQL functions on
TYPE:
|
class-attribute
is_shared: bool = True
property
provider: PostgresProvider
The bound PostgresProvider.
bind(
*, name: str, config: ConsecutiveCountConfig
) -> CircuitBreakerStrategy
Build a strategy for the named breaker and config.
Dispatches on the config.kind discriminator. Today only
consecutive_count is supported.
PostgresRateLimiterAdapter(
*,
provider: PostgresProvider | None = None,
env_prefix: str = "POSTGRES_",
prefix: str = "",
table_name: str = "grelmicro_rate_limiter",
auto_migrate: bool = True,
)
Bases: RateLimiterBackend
Postgres rate limiter adapter.
Wraps a PostgresProvider and supports both
TokenBucketConfig
and SlidingWindowConfig
algorithm configs via PL/pgSQL functions. Concurrent writes for
the same key are serialized with pg_advisory_xact_lock. Safe
across processes and machines.
Example:
from grelmicro.providers.postgres import PostgresProvider
from grelmicro.resilience import RateLimiter
from grelmicro.resilience.ratelimiter.postgres import (
PostgresRateLimiterAdapter,
)
async def main() -> None:
provider = PostgresProvider("postgresql://localhost:5432/app")
async with provider, PostgresRateLimiterAdapter(provider=provider):
rl = RateLimiter.token_bucket("api", capacity=10, refill_rate=1)
await rl.acquire(key="u1")
Read more in the Rate Limiter docs.
Initialize the rate limiter adapter.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
A pre-built
TYPE:
|
env_prefix
|
Environment variable prefix used by the implicit
TYPE:
|
prefix
|
Prefix prepended to every key the adapter writes. Use it to avoid collisions with other consumers of the same Postgres table.
TYPE:
|
table_name
|
Table that stores rate-limit state. Auto-created on
first connect (set
TYPE:
|
auto_migrate
|
When True (the default), the adapter creates the table
and SQL functions on
TYPE:
|
property
provider: PostgresProvider
The bound PostgresProvider.
bind(
config: TokenBucketConfig | SlidingWindowConfig,
) -> RateLimiterStrategy
Build a strategy for the given algorithm config.
Each strategy targets a pair of PL/pgSQL functions installed
on __aenter__. Functions read and write a single row of
{table_name} per key.
Bases: BaseModel
Random backoff: each delay is uniform random in [min_delay, max_delay].
Use this when you want bounded random spread without progressive growth. Common for cache-stampede protection (many clients miss the cache simultaneously: each waits a random short interval to avoid hammering the origin together).
For HTTP retries, prefer
ExponentialBackoff
with jitter. Random alone does not back off, so a persistent
failure retries at the same average rate forever.
Example:
from grelmicro.resilience import RandomBackoff, Retry
# Each retry waits a random 0.5-2.0 seconds
policy = Retry(
"stampede",
RandomBackoff(min_delay=0.5, max_delay=2.0),
when=CacheMissError,
attempts=3,
)
Read more in the Retry docs.
class-attribute
instance-attribute
kind: Literal['random'] = 'random'
Discriminator for the backoff Pydantic union.
class-attribute
instance-attribute
min_delay: PositiveFloat = 0.5
Minimum delay in seconds (inclusive).
class-attribute
instance-attribute
max_delay: PositiveFloat = 2.0
Maximum delay in seconds (inclusive).
RateLimiterRegistry(
source: Provider
| RateLimiterBackend
| type[Provider | RateLimiterBackend],
*,
name: str = "default",
)
RateLimiterBackend wrapper exposing (ratelimiter, name) registration.
Registered on a Grelmicro app via Grelmicro(uses=[RateLimiterRegistry(redis)]).
The active app resolves RateLimiter patterns to this Component's backend
on every call.
Accepts a Provider or a RateLimiterBackend. When given a Provider, the
component calls provider.ratelimiter() to build the matching adapter.
Example
from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import RateLimiter, RateLimiterRegistry
redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis, RateLimiterRegistry(redis)])
api = RateLimiter.token_bucket("api", capacity=10, refill_rate=1)
async with micro:
await api.acquire(key="user-1")
Initialize the Component with the wrapped backend.
| PARAMETER | DESCRIPTION |
|---|---|
source
|
A
TYPE:
|
name
|
Registration name. Multiple
TYPE:
|
class-attribute
kind: str = 'ratelimiter'
property
name: str
Return the registration name.
RateLimiter(
name: str,
config: RateLimiterConfig,
*,
backend: RateLimiterBackend | str | None = None,
)
Bases: Reconfigurable['RateLimiterConfig']
Rate limiter with a pluggable algorithm.
Most Python call sites should use the factory classmethods:
RateLimiter.token_bucket
for burst-friendly semantics or
RateLimiter.sliding_window 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
SlidingWindowConfig 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.
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
TYPE:
|
config
|
The algorithm configuration. Most callers should prefer the
Pass a
TYPE:
|
backend
|
An explicit backend instance. When Set this to skip the global registry, for example in tests or when running several backends at the same time.
TYPE:
|
property
name: str
Return the rate limiter identity.
property
backend: RateLimiterBackend
Bound rate-limiter backend, resolved on each call.
Resolution order:
1. An explicit backend= passed at construction wins.
2. The active Grelmicro app is consulted via
Grelmicro.current() so that micro.override(...) blocks
take effect.
| RAISES | DESCRIPTION |
|---|---|
OutOfContextError
|
No backend resolved in this scope. Pass
|
classmethod
from_config(
name: str,
config: RateLimiterConfig,
*,
backend: RateLimiterBackend | str | None = None,
) -> Self
Construct a RateLimiter from a name and a pre-built config.
Use this when configuration is assembled declaratively at
startup and the simple factory classmethods are not the right
fit. This declarative path opts out of live reload: the instance
is not addressable by ExternalConfig and stays on the config it
was built with.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the rate limiter instance.
TYPE:
|
config
|
The pre-built algorithm configuration.
TYPE:
|
backend
|
An explicit backend instance. When
TYPE:
|
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:
|
capacity
|
Maximum burst size. The bucket holds at most
TYPE:
|
refill_rate
|
Tokens replenished per second, up to
TYPE:
|
fail_open
|
When
TYPE:
|
backend
|
An explicit backend instance. When
TYPE:
|
classmethod
sliding_window(
name: str,
*,
limit: PositiveInt,
window: PositiveFloat,
fail_open: bool = False,
backend: RateLimiterBackend | str | None = None,
) -> Self
Construct a sliding-window rate limiter.
Convenience factory for the common case. Builds a
SlidingWindowConfig
internally and forwards to the constructor.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the rate limiter instance.
TYPE:
|
limit
|
Maximum number of requests allowed per window.
TYPE:
|
window
|
Window duration in seconds.
TYPE:
|
fail_open
|
When
TYPE:
|
backend
|
An explicit backend instance. When
TYPE:
|
async
acquire(
*, key: str = "default", 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). Defaults to
TYPE:
|
cost
|
Number of tokens to consume.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RateLimitResult
|
RateLimitResult with allowed, limit, remaining, |
RateLimitResult
|
retry_after, and reset_after fields. |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
async
acquire_or_raise(
*, key: str = "default", cost: int = 1
) -> RateLimitResult
Check rate limit, raise if exceeded.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier for rate limiting (e.g. IP address, user ID, session). Defaults to
TYPE:
|
cost
|
Number of tokens to consume.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RateLimitResult
|
RateLimitResult if allowed. |
| RAISES | DESCRIPTION |
|---|---|
RateLimitExceededError
|
If the rate limit is exceeded. |
async
allow(*, key: str = 'default', cost: int = 1) -> bool
Consume tokens and return whether the request is within the limit.
The boolean shortcut over acquire, for the common branch:
if await limiter.allow(key="user-1"):
... # served
else:
... # throttled
Use acquire instead when you need the retry_after or remaining
metadata on the deny branch.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier for rate limiting (e.g. IP address, user ID, session). Defaults to
TYPE:
|
cost
|
Number of tokens to consume.
TYPE:
|
async
peek(*, key: str = 'default') -> RateLimitResult
Check rate limit state without consuming tokens.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier for rate limiting (e.g. IP address, user ID, session). Defaults to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RateLimitResult
|
RateLimitResult reflecting the current state. |
RateLimitResult
|
|
RateLimitResult
|
succeed. |
async
reset(*, key: str = 'default') -> 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). Defaults to
TYPE:
|
async
wait(
*,
key: str = "default",
cost: int = 1,
max_wait: float | None = None,
) -> RateLimitResult
Wait until tokens are available, then consume them.
Polls acquire on the clock seam, sleeping retry_after
between attempts, until the request is admitted. A denied
acquire consumes nothing, so retrying is safe.
With max_wait set, gives up once the budget would be exceeded
and raises RateLimitExceededError. The default waits forever:
await limiter.wait(key="user-1")
result = await limiter.wait(key="user-1", cost=3, max_wait=2.0)
The wait runs on the clock seam, so VirtualClock drives it in
tests without real sleeping.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Identifier for rate limiting (e.g. IP address, user ID, session). Defaults to
TYPE:
|
cost
|
Number of tokens to consume.
TYPE:
|
max_wait
|
Maximum number of seconds to wait before giving up.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RateLimitResult
|
The allowed RateLimitResult once tokens are consumed. |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
RateLimitExceededError
|
If |
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(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
|
Algorithm configuration.
TYPE:
|
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.
async
acquire(*, key: str, cost: int) -> RateLimitResult
Try to acquire rate-limit tokens for key.
Returns a RateLimitResult with allowed, limit,
remaining, retry_after, and reset_after.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Rate-limit key (e.g. IP, user ID, session).
TYPE:
|
cost
|
Number of tokens to consume on this request.
TYPE:
|
async
peek(*, key: str) -> RateLimitResult
Return the current state for key without consuming tokens.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Rate-limit key to inspect.
TYPE:
|
async
reset(*, key: str) -> None
Delete rate-limit state for key, restoring full quota.
| PARAMETER | DESCRIPTION |
|---|---|
key
|
Rate-limit key to reset to full quota.
TYPE:
|
RateLimitExceededError(*, key: str, retry_after: float)
Bases: ResilienceError, AdmissionError
Rate limit exceeded error.
Raised when a rate limit check fails (too many requests).
Initialize the error.
instance-attribute
key = key
instance-attribute
retry_after = retry_after
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=
instance-attribute
allowed: bool
Whether the request is permitted.
instance-attribute
limit: int
Total quota (capacity for TokenBucketConfig, limit for SlidingWindowConfig).
instance-attribute
remaining: int
Remaining tokens or requests.
For algorithms with continuous state (SlidingWindowConfig,
GCRA-based strategies) this is an estimate rounded to the nearest
whole request. Enforcement still uses the exact internal state, so
the next acquire may be denied even when remaining > 0.
instance-attribute
retry_after: float
Seconds until the next request is allowed (0.0 if allowed).
instance-attribute
reset_after: float
Seconds until the full quota resets.
RedisCircuitBreakerAdapter(
*,
provider: RedisProvider | None = None,
env_prefix: str = "REDIS_",
prefix: str = "",
)
Bases: CircuitBreakerBackend
Redis circuit breaker adapter.
Builds a per-breaker
CircuitBreakerStrategy
that stores state in a Redis hash keyed {prefix}cb:{name}. All
admission and counter updates run as atomic Lua scripts so
concurrent replicas converge to the same state without
coordination locks.
Today the consecutive-count algorithm is the only strategy. Future
algorithm configs plug in through the same bind entry point.
last_error and last_error_time stay per-replica.
Example:
from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import CircuitBreakerRegistry, CircuitBreaker
redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis, CircuitBreakerRegistry(redis)])
payments = CircuitBreaker("payments")
Read more in the Circuit Breaker docs.
Initialize the circuit breaker adapter.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
A pre-built
TYPE:
|
env_prefix
|
Environment variable prefix used by the implicit
TYPE:
|
prefix
|
Prefix prepended to every Redis key the adapter writes. Use it to avoid collisions with other consumers of the same Redis database.
TYPE:
|
class-attribute
is_shared: bool = True
property
provider: RedisProvider
The bound RedisProvider.
bind(
*, name: str, config: ConsecutiveCountConfig
) -> CircuitBreakerStrategy
Build a strategy for the named breaker and config.
Dispatches on the config.kind discriminator. Today only
consecutive_count is supported.
RedisRateLimiterAdapter(
*,
provider: RedisProvider | None = None,
env_prefix: str = "REDIS_",
prefix: str = "",
)
Bases: RateLimiterBackend
Redis rate limiter adapter.
Wraps a RedisProvider and supports both
TokenBucketConfig
and SlidingWindowConfig
algorithm configs via atomic Lua scripts. Safe across processes
and machines.
Example:
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import RateLimiter
from grelmicro.resilience.ratelimiter.redis import RedisRateLimiterAdapter
async def main() -> None:
provider = RedisProvider("redis://localhost:6379/0")
async with RedisRateLimiterAdapter(provider=provider):
rl = RateLimiter.token_bucket("api", capacity=10, refill_rate=1)
await rl.acquire(key="u1")
Read more in the Rate Limiter docs.
Initialize the rate limiter adapter.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
A pre-built
TYPE:
|
env_prefix
|
Environment variable prefix used by the implicit
TYPE:
|
prefix
|
Prefix prepended to every Redis key the adapter writes. Use it to avoid collisions with other consumers of the same Redis database.
TYPE:
|
property
provider: RedisProvider
The bound RedisProvider.
bind(
config: TokenBucketConfig | SlidingWindowConfig,
) -> 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.
SQLiteRateLimiterAdapter(
*,
provider: SQLiteProvider | None = None,
env_prefix: str = "SQLITE_",
prefix: str = "",
table_name: str = "grelmicro_rate_limiter",
auto_migrate: bool = True,
)
Bases: RateLimiterBackend
SQLite rate limiter adapter.
Internal machinery. Most code should reach SQLite rate limiting
through a SQLiteProvider and RateLimiterRegistry(provider), not by
constructing this adapter directly. The adapter exists for expert
wiring and for the provider to build.
Borrows the connection and a shared lock from a SQLiteProvider
and supports both
TokenBucketConfig
and SlidingWindowConfig
algorithm configs. 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 same file. State
survives process restarts. For multi-replica coordination, use
RedisRateLimiterAdapter
or
PostgresRateLimiterAdapter.
Read more in the Rate Limiter docs.
Initialize the rate limiter adapter.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
A pre-built
TYPE:
|
env_prefix
|
Environment variable prefix used by the implicit
TYPE:
|
prefix
|
Prefix prepended to every key the adapter writes. Use it to avoid collisions with other consumers of the same SQLite table.
TYPE:
|
table_name
|
Table that stores rate-limit state. Auto-created on
first connect (set
TYPE:
|
auto_migrate
|
When True (the default), the adapter creates the table
on
TYPE:
|
property
provider: SQLiteProvider
The bound SQLiteProvider.
bind(
config: TokenBucketConfig | SlidingWindowConfig,
) -> RateLimiterStrategy
Build a strategy for the given algorithm config.
Bases: _BaseShieldConfig
Shield configuration for the api profile (the default).
Tuned for external HTTP APIs. Modest initial rate, generous timeouts, broad clamps. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['api'] = 'api'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 20
class-attribute
initial_max_rate: float = 2.0
class-attribute
adaptive_burst_capacity: float = 5.0
class-attribute
min_rate_floor: float = 0.25
class-attribute
initial_timeout: float = 10.0
class-attribute
timeout_clamp_min: float = 0.5
class-attribute
timeout_clamp_max: float = 60.0
class-attribute
backoff_scale: float = 1.0
class-attribute
backoff_cap: float = 30.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'api'
Bases: _BaseShieldConfig
Shield configuration for the internal profile.
Tuned for in-cluster RPC. High initial rate, short timeouts, tight budgets. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['internal'] = 'internal'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 10
class-attribute
initial_max_rate: float = 100.0
class-attribute
adaptive_burst_capacity: float = 200.0
class-attribute
min_rate_floor: float = 1.0
class-attribute
initial_timeout: float = 1.0
class-attribute
timeout_clamp_min: float = 0.05
class-attribute
timeout_clamp_max: float = 5.0
class-attribute
backoff_scale: float = 0.5
class-attribute
backoff_cap: float = 5.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'internal'
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(
error: ValidationError | str,
)
Retry(
name: str,
backoff: RetryBackoffConfig | None = None,
*,
when: WhenInput | None = None,
attempts: PositiveInt | None = None,
max_seconds: PositiveFloat | None = None,
config: RetryConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[RetryConfig]
Retry policy.
A named, reusable retry policy with three-paths configuration
and live reconfiguration. Use the
Retry.exponential
or Retry.constant
factory classmethods for the common case. Pass a pre-built
backoff config to the constructor when configuration is
assembled elsewhere.
Read more in the Retry docs.
Initialize the retry policy.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy. Used as the env namespace and exposed via the
TYPE:
|
backoff
|
The backoff algorithm config. Pass any
TYPE:
|
when
|
Outcome filter that engages the retry. Pass a
TYPE:
|
attempts
|
Total calls including the first. Default
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds. Retrying stops when either
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the retry policy identity.
classmethod
from_config(name: str, config: RetryConfig) -> Self
Construct a Retry from a name and a pre-built RetryConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
config
|
The pre-built retry 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:
|
classmethod
exponential(
name: str,
*,
when: WhenInput,
attempts: PositiveInt = 3,
max_seconds: PositiveFloat | None = None,
base_delay: PositiveFloat = 0.1,
max_delay: PositiveFloat = 30.0,
jitter: Literal[
"none", "full", "decorrelated"
] = "full",
env_load: bool | None = None,
) -> Self
Construct a retry policy with exponential backoff.
Convenience factory for the common case. Builds an
ExponentialBackoff
and forwards to the constructor.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
when
|
Outcome filter that engages the retry.
TYPE:
|
attempts
|
Total calls including the first.
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds, whichever comes first with
TYPE:
|
base_delay
|
Initial delay in seconds before the first retry.
TYPE:
|
max_delay
|
Maximum delay in seconds. Caps exponential growth.
TYPE:
|
jitter
|
Jitter mode.
TYPE:
|
env_load
|
Whether to read environment variables.
TYPE:
|
classmethod
constant(
name: str,
*,
when: WhenInput,
attempts: PositiveInt = 3,
max_seconds: PositiveFloat | None = None,
delay: PositiveFloat = 1.0,
env_load: bool | None = None,
) -> Self
Construct a retry policy with constant delay.
Convenience factory for polling-style retries. Builds a
ConstantBackoff
and forwards to the constructor.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
when
|
Outcome filter that engages the retry.
TYPE:
|
attempts
|
Total calls including the first.
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds, whichever comes first with
TYPE:
|
delay
|
Fixed delay in seconds between retries.
TYPE:
|
env_load
|
Whether to read environment variables.
TYPE:
|
RetryAttempt(
*,
number: int,
delay_before: float,
attempts: int,
matcher: Matcher,
loop: _AttemptLoop,
started_at: float,
max_seconds: float | None,
)
One iteration of a retry loop.
Yielded by Retry iteration and
by retrying. Used as an
async (or sync) context manager. Suppresses retryable
exceptions until attempts are exhausted, then re-raises the
underlying error with a PEP 678 note.
The block form sees only exceptions, not return values. Use the
decorator form (@retry or
policy(fn)) for result-based retry.
Initialize one retry attempt.
instance-attribute
number = number
1-indexed attempt number. The first call is 1.
instance-attribute
delay_before = delay_before
Seconds slept before this attempt (0.0 for the first).
Bases: BaseModel
Retry policy configuration.
Holds the top-level retry fields plus a discriminated backoff sub-config. Frozen Pydantic data class. Three-paths configuration: kwargs, instance, or env vars.
Read more in the Retry docs.
class-attribute
instance-attribute
attempts: PositiveInt = 3
Total calls including the first. attempts=1 means no retry. The default 3 allows two retries.
class-attribute
instance-attribute
max_seconds: PositiveFloat | None = None
Optional wall-clock budget in seconds, measured from the first attempt. Retrying stops as soon as either attempts is reached or this budget elapses, whichever comes first. The budget is checked between attempts, so one backoff may run slightly past it. None (default) means no time limit.
instance-attribute
when: Match
Outcome filter that engages the retry. Pass a Match (e.g. Match.exception(httpx.HTTPError) | Match.result(None)) or a shorthand: an exception class, a tuple of classes, or a predicate on the exception. BaseException subclasses outside Exception are never retried.
instance-attribute
backoff: RetryBackoffConfig
Backoff algorithm config. Discriminated union over ExponentialBackoff, ConstantBackoff, LinearBackoff, FibonacciBackoff, and RandomBackoff. Default: exponential with full jitter.
class-attribute
instance-attribute
model_config = {'arbitrary_types_allowed': True}
Bases: Protocol
A retry strategy for a specific backoff algorithm.
Built once per retry loop from a backoff config. The strategy holds any state the algorithm needs (for example the previous delay for decorrelated jitter) and computes one delay per upcoming attempt.
delay(attempt: int) -> float
Return the delay in seconds before retry attempt.
The strategy may apply jitter and clamp to its configured maximum.
| PARAMETER | DESCRIPTION |
|---|---|
attempt
|
Upcoming retry number.
TYPE:
|
Shield(
name: str,
config: _BaseShieldConfig | None = None,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
env_load: bool | None = None,
time_source: Callable[[], float] | None = None,
random_source: Callable[[], float] | None = None,
)
Bases: Reconfigurable[_BaseShieldConfig]
Shield resilience pattern.
Wraps a single async callable with:
- A per-attempt timeout estimated from the rolling p95 of the last 32 successful latencies.
- Exponential-jittered retries gated by a consecutive-failure budget.
- A CUBIC-style adaptive rate limiter that engages on the first slow-down and ramps back gradually.
- Optional cache and fallback recovery paths on give-up.
One Shield instance covers one logical dependency. Multiple
functions hitting the same dependency share one Shield and
therefore one retry budget and one CUBIC controller.
Read more in the Shield docs.
Initialize the Shield instance with the api profile by default.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Registration name of the Shield instance. Appears in logs, metrics, and PEP 678 notes attached on give-up.
TYPE:
|
config
|
A pre-built profile config (
TYPE:
|
timeout_errors
|
Exception classes treated as transient slow-downs.
TYPE:
|
max_rate
|
Optional hard ceiling on the adaptive bucket's rate in tokens per second.
TYPE:
|
cache
|
Optional cache instance read on give-up and written fire-and-forget on success.
TYPE:
|
cache_key
|
Optional callable returning the cache key for a call. Defaults to
TYPE:
|
fallback
|
Optional callable invoked on give-up when the cache path returns nothing. Receives the underlying exception.
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
time_source
|
Monotonic clock for tests. Defaults to
TYPE:
|
random_source
|
Uniform
TYPE:
|
property
name: str
Return the Shield instance name.
classmethod
from_config(name: str, config: _BaseShieldConfig) -> Self
Construct a Shield from a name and a pre-built profile config.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
config
|
The pre-built Shield profile configuration.
TYPE:
|
classmethod
internal(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the internal profile.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
classmethod
api(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the api profile (the default).
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
classmethod
slow(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the slow profile.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
async
run(
fn: Callable[..., Awaitable[Any]],
*args: Any,
**kwargs: Any,
) -> Any
Run fn(*args, **kwargs) through this Shield instance.
| PARAMETER | DESCRIPTION |
|---|---|
fn
|
Async callable to invoke under this Shield.
TYPE:
|
Bases: _BaseRateLimiterConfig
Precise sliding-window rate limiting.
Stores a single timestamp per key (about 72 bytes).
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 RateLimiter, SlidingWindowConfig
# 5 requests per 60-second sliding window.
rl = RateLimiter("auth", SlidingWindowConfig(limit=5, window=60))
Read more in the Rate Limiter docs.
class-attribute
instance-attribute
kind: Literal['sliding_window'] = 'sliding_window'
Discriminator for the algorithm Pydantic union.
instance-attribute
limit: PositiveInt
Maximum number of requests allowed per window.
instance-attribute
window: PositiveFloat
Window duration in seconds.
Bases: _BaseShieldConfig
Shield configuration for the slow profile.
Tuned for long-running calls: LLM inference, batch jobs, large queries. Low initial rate, large timeouts, tight failure budget. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['slow'] = 'slow'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 5
class-attribute
initial_max_rate: float = 0.5
class-attribute
adaptive_burst_capacity: float = 1.0
class-attribute
min_rate_floor: float = 0.05
class-attribute
initial_timeout: float = 120.0
class-attribute
timeout_clamp_min: float = 5.0
class-attribute
timeout_clamp_max: float = 600.0
class-attribute
backoff_scale: float = 2.0
class-attribute
backoff_cap: float = 60.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'slow'
Timeout(
name: str,
*,
seconds: PositiveFloat | None = None,
config: TimeoutConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[TimeoutConfig]
Timeout policy.
A named, reusable async deadline with three-paths configuration and live reconfiguration. Use as an async context manager or as a decorator on async functions.
Read more in the Timeout docs.
Initialize the timeout policy.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the timeout policy. Used as the env namespace and exposed via the
TYPE:
|
seconds
|
Deadline in seconds. Required unless
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the timeout policy identity.
classmethod
from_config(name: str, config: TimeoutConfig) -> Self
Construct a Timeout from a name and a pre-built TimeoutConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the timeout policy.
TYPE:
|
config
|
The pre-built timeout configuration.
TYPE:
|
Bases: BaseModel
Timeout policy configuration.
Frozen Pydantic data class. Three-paths configuration: kwargs, instance, or env vars.
Read more in the Timeout docs.
instance-attribute
seconds: PositiveFloat
Deadline in seconds. The inner block is cancelled and TimeoutError is raised when the deadline elapses.
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.
class-attribute
instance-attribute
kind: Literal['token_bucket'] = 'token_bucket'
Discriminator for the algorithm Pydantic union.
instance-attribute
capacity: PositiveInt
Maximum burst size. The bucket never holds more than capacity tokens.
instance-attribute
refill_rate: PositiveFloat
Tokens replenished per second, up to capacity.
falling_back(
*,
when: WhenInput,
default: Any = _UNSET,
factory: Callable[[BaseException], Any] | None = None,
) -> _FallbackBlock[Any]
Build a fallback context manager for the block form.
Use async with falling_back(...) as result: (or the sync
with form) to wrap a block of statements. Call
result.set(value) on the success path. On a matched
exception, the exception is suppressed and result.value
holds the configured default or factory output.
Exactly one of default= or factory= must be set.
| PARAMETER | DESCRIPTION |
|---|---|
when
|
Exception filter that engages the fallback.
TYPE:
|
default
|
Static fallback value. Mutually exclusive with factory.
TYPE:
|
factory
|
Callable producing the fallback value from the exception.
TYPE:
|
Fallback.
module-attribute
F = TypeVar('F', bound=Callable[..., Any])
Bases: BaseModel
Fallback policy configuration.
Holds the exception filter plus exactly one of default or
factory. Frozen Pydantic data class. Three-paths configuration:
kwargs, instance, or env vars.
Read more in the Fallback docs.
instance-attribute
when: Match
Exception filter that engages the fallback. Pass a Match or a shorthand: an exception class, a tuple of classes, or a predicate on the exception. BaseException subclasses outside Exception are never caught.
class-attribute
instance-attribute
default: Any = _UNSET
Static value returned when when matches. Mutually exclusive with factory. Exactly one must be set.
class-attribute
instance-attribute
factory: Callable[[BaseException], Any] | None = None
Callable that produces the fallback value from the exception. Mutually exclusive with default.
class-attribute
instance-attribute
model_config = {'arbitrary_types_allowed': True}
FallbackResult()
Holder for the value produced inside a falling_back block.
Call set(value)
on success. When the block raises an exception matching when=,
the exception is suppressed and the holder is filled with the
configured default (or the factory output). Access the resulting
value with the
value property
after the block exits.
Initialize an empty result holder.
property
value: T
Return the recorded value (success or fallback).
| RAISES | DESCRIPTION |
|---|---|
RuntimeError
|
When accessed before any value was set. |
set(value: T) -> None
Record the success value for the block.
Fallback(
name: str,
*,
when: WhenInput | None = None,
default: Any = _UNSET,
factory: Callable[[BaseException], Any] | None = None,
config: FallbackConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[FallbackConfig]
Fallback policy.
A named, reusable fallback policy with three-paths configuration
and live reconfiguration. Use the constructor for the common
case and Fallback.from_config when configuration is assembled
elsewhere.
Read more in the Fallback docs.
Initialize the fallback policy.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the fallback policy. Used as the env namespace and exposed via the
TYPE:
|
when
|
Exception filter that engages the fallback. Pass a
TYPE:
|
default
|
Static fallback value. Mutually exclusive with
TYPE:
|
factory
|
Callable that produces the fallback value from the matched exception. Mutually exclusive with
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the fallback policy identity.
classmethod
from_config(name: str, config: FallbackConfig) -> Self
Construct a Fallback from a name and a pre-built FallbackConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the fallback policy.
TYPE:
|
config
|
The pre-built fallback configuration.
TYPE:
|
fallback(
*,
when: WhenInput,
default: Any = _UNSET,
factory: Callable[[BaseException], Any] | None = None,
) -> Callable[[F], F]
Build an anonymous fallback decorator.
Exactly one of default= or factory= must be set.
| PARAMETER | DESCRIPTION |
|---|---|
when
|
Exception filter that engages the fallback.
TYPE:
|
default
|
Static fallback value. Mutually exclusive with factory.
TYPE:
|
factory
|
Callable producing the fallback value from the exception.
TYPE:
|
falling_back(
*,
when: WhenInput,
default: Any = _UNSET,
factory: Callable[[BaseException], Any] | None = None,
) -> _FallbackBlock[Any]
Build a fallback context manager for the block form.
Use async with falling_back(...) as result: (or the sync
with form) to wrap a block of statements. Call
result.set(value) on the success path. On a matched
exception, the exception is suppressed and result.value
holds the configured default or factory output.
Exactly one of default= or factory= must be set.
| PARAMETER | DESCRIPTION |
|---|---|
when
|
Exception filter that engages the fallback.
TYPE:
|
default
|
Static fallback value. Mutually exclusive with factory.
TYPE:
|
factory
|
Callable producing the fallback value from the exception.
TYPE:
|
Retry.
module-attribute
logger = getLogger(__name__)
module-attribute
F = TypeVar('F', bound=Callable[..., Any])
module-attribute
WhenInput = (
Match
| type[Exception]
| tuple[type[Exception], ...]
| Callable[[Exception], bool]
)
User-facing shape accepted by when=.
A Match instance, or one of the
shorthand forms a Match would build for you: a single exception
class, a tuple of classes, or a callable predicate on the
exception. Bare shapes are coerced to Match.exception(...).
module-attribute
retry = _RetryFactory()
module-attribute
retrying = _RetryingFactory()
Bases: BaseModel
Retry policy configuration.
Holds the top-level retry fields plus a discriminated backoff sub-config. Frozen Pydantic data class. Three-paths configuration: kwargs, instance, or env vars.
Read more in the Retry docs.
class-attribute
instance-attribute
attempts: PositiveInt = 3
Total calls including the first. attempts=1 means no retry. The default 3 allows two retries.
class-attribute
instance-attribute
max_seconds: PositiveFloat | None = None
Optional wall-clock budget in seconds, measured from the first attempt. Retrying stops as soon as either attempts is reached or this budget elapses, whichever comes first. The budget is checked between attempts, so one backoff may run slightly past it. None (default) means no time limit.
instance-attribute
when: Match
Outcome filter that engages the retry. Pass a Match (e.g. Match.exception(httpx.HTTPError) | Match.result(None)) or a shorthand: an exception class, a tuple of classes, or a predicate on the exception. BaseException subclasses outside Exception are never retried.
instance-attribute
backoff: RetryBackoffConfig
Backoff algorithm config. Discriminated union over ExponentialBackoff, ConstantBackoff, LinearBackoff, FibonacciBackoff, and RandomBackoff. Default: exponential with full jitter.
class-attribute
instance-attribute
model_config = {'arbitrary_types_allowed': True}
RetryAttempt(
*,
number: int,
delay_before: float,
attempts: int,
matcher: Matcher,
loop: _AttemptLoop,
started_at: float,
max_seconds: float | None,
)
One iteration of a retry loop.
Yielded by Retry iteration and
by retrying. Used as an
async (or sync) context manager. Suppresses retryable
exceptions until attempts are exhausted, then re-raises the
underlying error with a PEP 678 note.
The block form sees only exceptions, not return values. Use the
decorator form (@retry or
policy(fn)) for result-based retry.
Initialize one retry attempt.
instance-attribute
number = number
1-indexed attempt number. The first call is 1.
instance-attribute
delay_before = delay_before
Seconds slept before this attempt (0.0 for the first).
Retry(
name: str,
backoff: RetryBackoffConfig | None = None,
*,
when: WhenInput | None = None,
attempts: PositiveInt | None = None,
max_seconds: PositiveFloat | None = None,
config: RetryConfig | None = None,
env_load: bool | None = None,
)
Bases: Reconfigurable[RetryConfig]
Retry policy.
A named, reusable retry policy with three-paths configuration
and live reconfiguration. Use the
Retry.exponential
or Retry.constant
factory classmethods for the common case. Pass a pre-built
backoff config to the constructor when configuration is
assembled elsewhere.
Read more in the Retry docs.
Initialize the retry policy.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy. Used as the env namespace and exposed via the
TYPE:
|
backoff
|
The backoff algorithm config. Pass any
TYPE:
|
when
|
Outcome filter that engages the retry. Pass a
TYPE:
|
attempts
|
Total calls including the first. Default
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds. Retrying stops when either
TYPE:
|
config
|
A pre-built
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
property
name: str
Return the retry policy identity.
classmethod
from_config(name: str, config: RetryConfig) -> Self
Construct a Retry from a name and a pre-built RetryConfig.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
config
|
The pre-built retry 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:
|
classmethod
exponential(
name: str,
*,
when: WhenInput,
attempts: PositiveInt = 3,
max_seconds: PositiveFloat | None = None,
base_delay: PositiveFloat = 0.1,
max_delay: PositiveFloat = 30.0,
jitter: Literal[
"none", "full", "decorrelated"
] = "full",
env_load: bool | None = None,
) -> Self
Construct a retry policy with exponential backoff.
Convenience factory for the common case. Builds an
ExponentialBackoff
and forwards to the constructor.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
when
|
Outcome filter that engages the retry.
TYPE:
|
attempts
|
Total calls including the first.
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds, whichever comes first with
TYPE:
|
base_delay
|
Initial delay in seconds before the first retry.
TYPE:
|
max_delay
|
Maximum delay in seconds. Caps exponential growth.
TYPE:
|
jitter
|
Jitter mode.
TYPE:
|
env_load
|
Whether to read environment variables.
TYPE:
|
classmethod
constant(
name: str,
*,
when: WhenInput,
attempts: PositiveInt = 3,
max_seconds: PositiveFloat | None = None,
delay: PositiveFloat = 1.0,
env_load: bool | None = None,
) -> Self
Construct a retry policy with constant delay.
Convenience factory for polling-style retries. Builds a
ConstantBackoff
and forwards to the constructor.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the retry policy.
TYPE:
|
when
|
Outcome filter that engages the retry.
TYPE:
|
attempts
|
Total calls including the first.
TYPE:
|
max_seconds
|
Optional wall-clock budget in seconds, whichever comes first with
TYPE:
|
delay
|
Fixed delay in seconds between retries.
TYPE:
|
env_load
|
Whether to read environment variables.
TYPE:
|
Shield resilience pattern.
module-attribute
shield = _ShieldDecorator()
module-attribute
ShieldConfig = (
InternalShieldConfig
| ApiShieldConfig
| SlowShieldConfig
)
Discriminated union over the three Shield profile configs.
Bases: _BaseShieldConfig
Shield configuration for the api profile (the default).
Tuned for external HTTP APIs. Modest initial rate, generous timeouts, broad clamps. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['api'] = 'api'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 20
class-attribute
initial_max_rate: float = 2.0
class-attribute
adaptive_burst_capacity: float = 5.0
class-attribute
min_rate_floor: float = 0.25
class-attribute
initial_timeout: float = 10.0
class-attribute
timeout_clamp_min: float = 0.5
class-attribute
timeout_clamp_max: float = 60.0
class-attribute
backoff_scale: float = 1.0
class-attribute
backoff_cap: float = 30.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'api'
Bases: _BaseShieldConfig
Shield configuration for the internal profile.
Tuned for in-cluster RPC. High initial rate, short timeouts, tight budgets. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['internal'] = 'internal'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 10
class-attribute
initial_max_rate: float = 100.0
class-attribute
adaptive_burst_capacity: float = 200.0
class-attribute
min_rate_floor: float = 1.0
class-attribute
initial_timeout: float = 1.0
class-attribute
timeout_clamp_min: float = 0.05
class-attribute
timeout_clamp_max: float = 5.0
class-attribute
backoff_scale: float = 0.5
class-attribute
backoff_cap: float = 5.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'internal'
Shield(
name: str,
config: _BaseShieldConfig | None = None,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
env_load: bool | None = None,
time_source: Callable[[], float] | None = None,
random_source: Callable[[], float] | None = None,
)
Bases: Reconfigurable[_BaseShieldConfig]
Shield resilience pattern.
Wraps a single async callable with:
- A per-attempt timeout estimated from the rolling p95 of the last 32 successful latencies.
- Exponential-jittered retries gated by a consecutive-failure budget.
- A CUBIC-style adaptive rate limiter that engages on the first slow-down and ramps back gradually.
- Optional cache and fallback recovery paths on give-up.
One Shield instance covers one logical dependency. Multiple
functions hitting the same dependency share one Shield and
therefore one retry budget and one CUBIC controller.
Read more in the Shield docs.
Initialize the Shield instance with the api profile by default.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Registration name of the Shield instance. Appears in logs, metrics, and PEP 678 notes attached on give-up.
TYPE:
|
config
|
A pre-built profile config (
TYPE:
|
timeout_errors
|
Exception classes treated as transient slow-downs.
TYPE:
|
max_rate
|
Optional hard ceiling on the adaptive bucket's rate in tokens per second.
TYPE:
|
cache
|
Optional cache instance read on give-up and written fire-and-forget on success.
TYPE:
|
cache_key
|
Optional callable returning the cache key for a call. Defaults to
TYPE:
|
fallback
|
Optional callable invoked on give-up when the cache path returns nothing. Receives the underlying exception.
TYPE:
|
env_load
|
Whether to read environment variables. Defaults to the process-wide
TYPE:
|
time_source
|
Monotonic clock for tests. Defaults to
TYPE:
|
random_source
|
Uniform
TYPE:
|
property
name: str
Return the Shield instance name.
classmethod
from_config(name: str, config: _BaseShieldConfig) -> Self
Construct a Shield from a name and a pre-built profile config.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
config
|
The pre-built Shield profile configuration.
TYPE:
|
classmethod
internal(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the internal profile.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
classmethod
api(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the api profile (the default).
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
classmethod
slow(
name: str,
*,
timeout_errors: tuple[type[BaseException], ...]
| None = None,
max_rate: PositiveFloat | None = None,
cache: Any = None,
cache_key: Callable[..., str] | None = None,
fallback: Callable[[BaseException], Any]
| Callable[[BaseException], Awaitable[Any]]
| None = None,
) -> Self
Construct a Shield with the slow profile.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Name of the Shield instance.
TYPE:
|
async
run(
fn: Callable[..., Awaitable[Any]],
*args: Any,
**kwargs: Any,
) -> Any
Run fn(*args, **kwargs) through this Shield instance.
| PARAMETER | DESCRIPTION |
|---|---|
fn
|
Async callable to invoke under this Shield.
TYPE:
|
Bases: _BaseShieldConfig
Shield configuration for the slow profile.
Tuned for long-running calls: LLM inference, batch jobs, large queries. Low initial rate, large timeouts, tight failure budget. The profile parameters are frozen.
Read more in the Shield docs.
class-attribute
instance-attribute
kind: Literal['slow'] = 'slow'
Discriminator for the Shield profile Pydantic union.
class-attribute
max_consecutive_failures: int = 5
class-attribute
initial_max_rate: float = 0.5
class-attribute
adaptive_burst_capacity: float = 1.0
class-attribute
min_rate_floor: float = 0.05
class-attribute
initial_timeout: float = 120.0
class-attribute
timeout_clamp_min: float = 5.0
class-attribute
timeout_clamp_max: float = 600.0
class-attribute
backoff_scale: float = 2.0
class-attribute
backoff_cap: float = 60.0
class-attribute
max_rate_cap_default: float | None = None
class-attribute
profile_name: str = 'slow'