Resilix

A declarative execution-policy layer for Python functions. It provides a small decorator DSL for describing how a callable should behave on failure — retries, backoff, fallbacks, result reshaping, timeouts, a circuit breaker, and lifecycle hooks — while keeping the function body itself free of control-flow noise.

• Zero dependencies; standard library only.
• Python 3.10+.
• One policy model works identically on def and async def.
• Typed (py.typed, ParamSpec/TypeVar): decorated callables keep their original signature and return type.
• Invalid configuration fails at declaration time, not at runtime.

from resilix import protect

@protect.when(TimeoutError, retry=3, backoff="exponential")
def fetch(url): ...

Installation

pip install resilix

From a checkout, for development:

pip install -e ".[dev]"     # with the test/type-check toolchain

Model

Resilix is built from five orthogonal concepts. A new kind of reaction is a new Trigger; the engine does not change.

Event  →  Trigger  →  Rule  →  Policy  →  Engine

Concept	Meaning
Event	A snapshot of one attempt: ResultEvent or ExceptionEvent.
Trigger	Decides whether an event is interesting.
Rule	One trigger bound to one action bundle.
Policy	An ordered, immutable set of rules; composed with |.
Engine	Runs the target, evaluates rules, retries or terminates.

The protect decorator

protect is a stateless facade. Each method returns a decorator. Stacking several protect decorators on one function merges them into a single flat policy (no nested call frames). Stacking never mutates an existing wrapper: wrap(f, p) returns a new wrapper around the original target, and the policy is compiled once at decoration time (no per-call lookups).

A policy has orthogonal axes:

Axis	Methods
reactions	when, when_result, on (custom trigger)
limits	timeout, limits
guards	bulkhead, locked, single_flight, circuit
executor	offload, isolate (at most one per policy)
hooks	always, observe
code block	suppress (context manager)

Decorator	Purpose
protect.when(on, **actions)	React to an exception type (or a tuple of types).
protect.when_result(predicate, **actions)	React to a returned value that satisfies predicate.
protect.always(cb)	Run cb exactly once after the whole operation, on success or failure.
protect.on(trigger, **actions)	React to a custom Trigger (any object with matches(event)).
protect.observe(*, on_start=None, on_finish=None)	Lifecycle hooks receiving a CallInfo.
protect.limits(*, timeout=None, deadline=None, cancel_grace=None, allow_sync=False)	Per-attempt timeout and total-budget deadline.
protect.timeout(seconds, *, deadline=None, cancel_grace=None, allow_sync=False)	Cap a single attempt; a slow call becomes TimeoutError.
protect.circuit(*, failures=5, reset_timeout=30.0, on=Exception, failure_when=None, on_state_change=None, half_open_max=1, window=None)	Wrap the function in a circuit breaker.
protect.bulkhead(limit, *, key=None, wait=True)	At most limit concurrent calls (per key(...)).
protect.locked(lock, *, allow_retry=False)	Serialise calls under a lock object or self.<attr>.
protect.single_flight(*, key=None)	async: concurrent calls with one key share one execution.
protect.offload(pool="default", *, workers=None, abandon_after=None)	Run a sync function in a named thread pool; the wrapper becomes async.
protect.isolate(timeout, *, workers=None, name=None)	Run a sync function in a child process; crash/hang → IsolatedTaskError.
protect.suppress(*exceptions, log=None, message=None, timeout=None, notify=None)	Context manager for a block of code.
protect.methods(policy, *, include=None, exclude=(), circuit=None)	Apply one policy across a class's methods.

Actions

when and when_result accept the same keyword actions. A rule allows retry and pacing options plus exactly one terminal action.

Keyword	Effect
retry	Total attempts, including the first (int or a Retry).
delay	Base pause between attempts (float, or delay(exc) -> float).
backoff	Growth curve: "constant", "linear", "exponential".
jitter	Multiply the pause by a random factor in [0.5, 1.5].
max_delay	Ceiling applied after backoff and jitter.
mutate	mutate(kwargs, error) run before each retry; edits in place or returns a dict.
notify	Called with an Event on every matched attempt.
log	Built-in logging: a logger (logging/loguru), its bound method, or any callable(str).
message	Prefix for the log line (requires log).
return_	Terminal. Return a fixed value (including None).
fallback	Terminal. Try one or more alternatives; return the first success.
transform	Terminal. Reshape the outcome: fn(result) or fn(exc).
raise_	Terminal. Raise a new exception; the original is kept as __cause__.

from resilix import protect

# Retry, then re-raise the last exception.
@protect.when(TimeoutError, retry=3)
def ping(): ...

# Exponential backoff with jitter and a ceiling.
@protect.when(ConnectionError, retry=5, backoff="exponential", jitter=True, max_delay=10)
def call(): ...

# Fall back to alternatives, then a fixed default.
@protect.when(ConnectionError, fallback=[replica, cache], return_={})
def read(key): ...

# React to a value rather than an exception.
@protect.when_result(lambda r: r.status_code == 429, retry=3, delay=1.0, return_="rate_limited")
def request(): ...

# Wrap a low-level error, preserving the cause.
@protect.when(OSError, raise_=lambda exc: StorageError("io failed"))
def save(row): ...

# Run a finaliser once, whatever happens.
@protect.always(close_connection)
def query(): ...

# Log the reaction without a hand-written notify lambda.
@protect.when(ConnectionError, retry=3, delay=1.0, log=logger.warning, message="pg connect")
async def connect(): ...
# -> "pg connect: ConnectionError: refused (attempt 1/3); retrying in 1.00s"

Events

Every notify hook receives an ExceptionEvent or a ResultEvent with the attempt and the engine's decision: attempt, max_attempts, will_retry, next_delay (pause before the next attempt, None when giving up) and elapsed (seconds since the first attempt; None when the engine did not measure time).

Logging

log= / message= on when, when_result, on, suppress, gather and spawn, or an explicit hook via log_to(target, message="", *, level="warning"):

from resilix import log_to

@protect.when(ValueError, return_=None, notify=log_to(logger, "parse", level="error"))
def parse(raw): ...

For error/critical/exception levels the traceback is attached (exc_info for logging, opt(exception=...) for loguru — no loguru dependency is required).

Custom triggers

Any object with matches(event) -> bool (and an optional specificity: int) is a trigger. The engine calls it with an ExceptionEvent on errors and a ResultEvent on results:

class Retryable:
    specificity = 0
    def matches(self, event):
        return getattr(getattr(event, "exception", None), "retryable", False)

@protect.on(Retryable(), retry=3, delay=0.5)
def call(): ...

Blocks of code: suppress

The most common service pattern — try: ... except Exception: logger.warning(...) around a few lines inside a bigger function — is a context manager:

with protect.suppress(Exception, log=logger.warning, message="ACK failed"):
    msg.ack()

async with protect.suppress(Exception, log=logger.warning, message="cleanup", timeout=5):
    await cleanup()

with protect.suppress(KeyError) as s:
    value = data["key"]
if s.error is not None: ...

exceptions defaults to Exception. timeout= requires async with (a block exceeding it raises TimeoutError, swallowed only if it is in exceptions). KeyboardInterrupt, SystemExit and asyncio.CancelledError are never swallowed. Create a new suppress(...) per use.

Policies

A Policy is an ordered, immutable set of rules. Build one directly, use it as a decorator, and combine policies with |. On composition the most specific exception (deepest in the MRO) wins a tie; time limits merge to the strictest (smallest) value; always finalisers run in sequence.

from resilix import Policy, compose

network = Policy(on=TimeoutError, retry=3, backoff="exponential")
missing = Policy(on=KeyError, return_=None)

resilient = network | missing        # or compose(network, missing)

@resilient
def get(user_id): ...

compose(*policies) and all_of(iterable) are thin helpers over |.

Composition rules: reactions — by MRO specificity; guards and observers — in declaration order (top decorator first); limits — the strictest value (cancel_grace — the largest); executor and single_flight — at most one, otherwise PolicyConfigurationError.

Retry

Retry is a reusable descriptor for the pacing of a policy.

from resilix import Retry, Policy

backoff = Retry(attempts=5, delay=0.2, backoff="exponential", jitter=True, max_delay=10)

@Policy(on=OSError, retry=backoff, fallback=from_disk)
def load(path): ...

attempts is the total number of attempts including the first (attempts=1 never retries). Retry is frozen, so one object is safe to share across a module or service. BackoffStrategy is the literal type of the backoff field. Invalid values (attempts < 1, negative delay, unknown backoff) raise RetryConfigurationError on construction.

Circuit breaker

CircuitBreaker is a thread-safe decorator with its own state, separate from Policy. After failures consecutive matching errors the circuit opens and rejects calls with CircuitOpenError for reset_timeout seconds, then admits exactly half_open_max (default 1) half-open trials — concurrent callers beyond that get CircuitOpenError. A successful trial closes the circuit, a failure reopens it.

Options: failure_when=pred (count a result as a failure, e.g. HTTP 503), on_state_change=cb(old, new) (metrics/alerts), window=seconds (count failures in a sliding window instead of consecutively). Manual mode without a decorator: if breaker.allow(): ... breaker.record_success() / breaker.record_failure() (breaker.release() for a neutral outcome).

from resilix import CircuitBreaker

breaker = CircuitBreaker(failures=5, reset_timeout=30, on=ConnectionError)

@breaker
def read():  ...

@breaker
def write(x): ...

breaker.state        # "closed" | "open" | "half_open"

Use one breaker object across several functions to give them shared state. protect.circuit(...) is a convenience wrapper that constructs a breaker for a single function.

Class-level policies

protect.methods(policy, ...) applies one policy across a class's methods.

from resilix import Policy, CircuitBreaker, protect

io = Policy(on=(TimeoutError, ConnectionError), retry=3, backoff="exponential")

@protect.methods(io, exclude=("ping",))
class StorageClient:
    def read(self, key): ...
    def write(self, key, val): ...
    def ping(self): ...          # left unprotected
    def _connect(self): ...      # private — unprotected by default

Selection:

• By default only local, public methods (defined on the class itself, not inherited) are wrapped.
• Dunder names (__x__), private names (_x), and descriptors (property, classmethod, staticmethod) are skipped unless named in include.
• include=(...) is an exact whitelist that overrides the public-only default.
• exclude=(...) subtracts from the default.
• A name in include/exclude that is not defined on the class raises PolicyConfigurationError at declaration time.

Passing circuit=CircuitBreaker(...) additionally wraps each selected method in the same breaker, so its state is shared across methods and instances. The layering is circuit(policy(method)), so retries inside the policy count as a single failure to the breaker. The decorator returns the same class object; it does not create a subclass.

Concurrency guards

@protect.bulkhead(4, key=lambda tenant, *_: tenant, wait=False)   # BulkheadFullError when full
async def crawl(tenant, url): ...

class TokenClient:
    def __init__(self):
        self._lock = asyncio.Lock()

    @protect.locked("_lock")                     # lock taken from self._lock
    async def refresh(self): ...

@protect.single_flight()                         # concurrent loads of one key share a result
async def load_config(name): ...

• bulkhead creates async semaphores per event loop; keyed semaphores are dropped when unused.
• locked checks the lock type (asyncio.Lock for async def, a threading lock for def) and refuses retry unless allow_retry=True (the lock is held for the whole operation, retry pauses included).
• single_flight is not a cache: once the call finishes the key is released.

Executors

@protect.offload("parse", workers=4)        # sync def -> async wrapper over a named thread pool
def parse_html(html): ...
tree = await parse_html(html)

@protect.isolate(timeout=20, workers=2)     # child process; SIGSEGV/hang -> IsolatedTaskError
def extract(html): ...

isolate needs a module-level function (importable by the spawn child) and picklable arguments/results. In the child the raw function runs (no recursive isolation). A crashed worker is retried once, a hung one is killed; the pool is restarted and the service keeps running. Shut pools down with resilix.shutdown_thread_pools() / resilix.shutdown_isolated_pools().

Observability

@protect.observe(on_start=lambda info: ..., on_finish=lambda info: metrics.observe(info.elapsed))
@protect.when(ConnectionError, retry=3)
def call(): ...

CallInfo carries func, args, kwargs, attempts, elapsed, result, exception, ok and events (the event of every attempt).

Task groups

results, errors = await resilix.gather(
    *(functools.partial(fetch, url) for url in urls),
    limit=10, policy=io_policy, log=logger.warning, message="fetch",
)
# results: values in input order (None where an item failed); errors: [(index, exc)]

resilix.spawn(refresh_cache(), name="refresh", log=logger.error)   # strong ref + logged failure
await resilix.shutdown_spawned()                                     # cancel and await on shutdown

With policy= items must be zero-argument callables (a coroutine object cannot be retried). Cancellation is never turned into an item error.

Async

The same policies apply to coroutines. Delays use asyncio.sleep; predicates, transforms, mutations, and fallbacks may be sync or async. asyncio.CancelledError is always propagated and cannot be swallowed by any rule.

@protect.when(TimeoutError, retry=3, delay=0.1, fallback=from_cache)
async def fetch(url): ...

Objects with async def __call__ and functools.partial of a coroutine function are recognised as async targets. Fallback candidates may themselves be protected functions.

Behavioural contracts

• retry=N counts the first call: retry=3 is up to three attempts total, retry=1 runs once and never retries.
• when_result(..., retry=N) requires a terminal action (return_, fallback, transform, or raise_); otherwise it raises PolicyConfigurationError at declaration time.
• delay / backoff / jitter / max_delay require retry; using them alone raises PolicyConfigurationError.
• rethrow= is a deprecated synonym of raise_= and emits DeprecationWarning.
• protect.timeout differs by execution model: the async engine cancels the coroutine (and with cancel_grace= bounds how long it waits for the cancelled coroutine's cleanup). On a sync function a timeout cannot stop the call — the thread keeps running in the background — so it is a PolicyConfigurationError unless you opt in with allow_sync=True. Prefer protect.offload(...) or protect.isolate(...). The worker thread receives a copy of the caller's contextvars.
• deadline is a budget: a retry pause that does not fit into the remaining budget is not taken (the rule's terminal action runs immediately), and for async functions each attempt is also bounded by the remaining budget.
• Generator and async-generator functions are rejected with PolicyConfigurationError (the decorator would only cover creating the generator, not iterating it). protect.methods skips them by default.
• CircuitBreaker is not part of Policy: it cannot be merged with |, has no return_/fallback, and stacks as a nested decorator.
• A failing hook (notify, log, always, observe, on_state_change) does not break the call, but emits a ResilixHookWarning (once per hook); in a fallback chain only the last error propagates.
• KeyboardInterrupt, SystemExit and asyncio.CancelledError are never swallowed — not by rules, suppress, gather or spawn.
• protect.methods does not follow inheritance: a method overridden in a subclass is unprotected. notify hooks see self as event.args[0].

A rendered guide with side-by-side examples is in docs/usage.html.

Public API

Everything below is exported from the top-level resilix package.

• Decorators / policies: protect, Policy, compose, all_of.
• Retry: Retry, BackoffStrategy.
• Circuit breaker: CircuitBreaker.
• Rules & triggers: Rule, build_rule, Trigger, ExceptionTrigger, ResultTrigger.
• Events: Event, ExceptionEvent, ResultEvent, CallInfo.
• Block / logging / hooks: Suppress, LogHook, log_to, Observer.
• Task groups: gather, spawn, spawned_tasks, shutdown_spawned.
• Executors: shutdown_thread_pools, shutdown_isolated_pools.
• Sentinel: UNSET (distinguishes "return None" from "no value set").
• Errors: ResilixError (base), PolicyConfigurationError, RetryConfigurationError, AsyncCallbackInSyncContextError, PolicyExecutionError, CircuitOpenError, BulkheadFullError, IsolatedTaskError; warning ResilixHookWarning.

Error	Raised when
PolicyConfigurationError	A policy is declared with impossible or conflicting options.
RetryConfigurationError	A Retry is built with invalid parameters.
AsyncCallbackInSyncContextError	An async callback is attached to a synchronous target.
PolicyExecutionError	The engine reached an impossible state at runtime.
CircuitOpenError	A call was rejected because the circuit is open (or the half-open slot is taken).
BulkheadFullError	A bulkhead(wait=False) had no free slot.
IsolatedTaskError	An isolated task crashed, hung, or the pool was unavailable (reason).
ResilixHookWarning	A hook raised; the call continued.

Performance

Policies without retries, limits, guards and operation hooks (e.g. @protect.when(KeyError, return_=None)) use a fast path: one try/except <handled types> around the call, no kwargs copy, no clock, no loop. On CPython 3.11 the success-path overhead dropped from ~0.68 µs to ~0.12 µs per call (bare call ~0.05 µs).

Development

pip install -e ".[dev]"
pytest            # unit, sync, async, composition, security, class-level
mypy              # strict typing

Runnable examples are in examples/; adversarial and concurrency tests are in tests/security/.

License

MIT.