Skip to content

API

pico_resilience.decorators

Marker decorators: store the policy on the function and attach the matching interceptor via intercepted_by (same idiom as pico-pydantic).

retryable(_func=None, *, max_attempts=3, backoff_seconds=0.1, retry_on=(Exception,))

Retry on failure with exponential backoff. Sync or async. Raises RetryExhaustedError when every attempt failed.

Source code in src/pico_resilience/decorators.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def retryable(
    _func: Callable | None = None,
    *,
    max_attempts: int = 3,
    backoff_seconds: float = 0.1,
    retry_on: Tuple[Type[BaseException], ...] = (Exception,),
):
    """Retry on failure with exponential backoff. Sync or async. Raises
    ``RetryExhaustedError`` when every attempt failed."""
    if max_attempts < 1:
        raise ValueError("max_attempts must be >= 1")

    def dec(fn):
        setattr(
            fn, RETRY_META, {"max_attempts": max_attempts, "backoff_seconds": backoff_seconds, "retry_on": retry_on}
        )
        from pico_ioc import intercepted_by

        from .interceptors import RetryInterceptor

        return intercepted_by(RetryInterceptor)(fn)

    return dec(_func) if callable(_func) else dec

circuit_breaker(_func=None, *, failure_threshold=5, reset_timeout_seconds=30.0)

Per-method circuit: opens after N consecutive failures (calls raise CircuitOpenError), half-opens after the reset timeout.

Source code in src/pico_resilience/decorators.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def circuit_breaker(
    _func: Callable | None = None,
    *,
    failure_threshold: int = 5,
    reset_timeout_seconds: float = 30.0,
):
    """Per-method circuit: opens after N consecutive failures (calls raise
    ``CircuitOpenError``), half-opens after the reset timeout."""
    if failure_threshold < 1:
        raise ValueError("failure_threshold must be >= 1")

    def dec(fn):
        setattr(
            fn, CIRCUIT_META, {"failure_threshold": failure_threshold, "reset_timeout_seconds": reset_timeout_seconds}
        )
        from pico_ioc import intercepted_by

        from .interceptors import CircuitBreakerInterceptor

        return intercepted_by(CircuitBreakerInterceptor)(fn)

    return dec(_func) if callable(_func) else dec

timeout(seconds)

Bound an async method with asyncio.timeout. Sync methods are rejected at import time — a thread cannot be cancelled, so a sync timeout would lie.

Source code in src/pico_resilience/decorators.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def timeout(seconds: float):
    """Bound an async method with ``asyncio.timeout``. Sync methods are
    rejected at import time — a thread cannot be cancelled, so a sync
    timeout would lie."""
    if not isinstance(seconds, (int, float)) or seconds <= 0:
        raise ValueError("timeout(seconds) requires a positive number")

    def dec(fn):
        if not inspect.iscoroutinefunction(fn):
            raise TypeError(f"@timeout only supports async methods; {fn.__qualname__} is sync")
        setattr(fn, TIMEOUT_META, {"seconds": float(seconds)})
        from pico_ioc import intercepted_by

        from .interceptors import TimeoutInterceptor

        return intercepted_by(TimeoutInterceptor)(fn)

    return dec

pico_resilience.interceptors

Resilience interceptors (pico-ioc MethodInterceptor singletons).

Chain rule: retry attempts after the first re-invoke the method directly, so interceptors inside retry's position are skipped on retries. Combine policies with @retryable as the TOP decorator (it attaches last and runs innermost): @cacheable/@circuit_breaker below it wrap the whole retry loop — the cache stores the final result and the circuit counts exhausted retries, not individual attempts.

RetryInterceptor

Retries the method per its @retryable policy.

Source code in src/pico_resilience/interceptors.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@component(scope="singleton")
class RetryInterceptor:
    """Retries the method per its ``@retryable`` policy."""

    def __init__(self, settings: ResilienceSettings):
        self.settings = settings

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        meta = _meta(ctx, RETRY_META)
        if not self.settings.enabled or not meta:
            return call_next(ctx)
        if _is_async(ctx):
            return self._run_async(ctx, call_next, meta)
        return self._run_sync(ctx, call_next, meta)

    def _run_sync(self, ctx, call_next, meta):
        last: Exception | None = None
        for attempt in range(1, meta["max_attempts"] + 1):
            try:
                return call_next(ctx) if attempt == 1 else ctx.method(*ctx.args, **ctx.kwargs)
            except meta["retry_on"] as exc:
                last = exc
                self._log(ctx, attempt, meta, exc)
                if attempt < meta["max_attempts"]:
                    time.sleep(self._delay(meta, attempt))
        raise RetryExhaustedError(ctx.name, meta["max_attempts"], last)

    async def _run_async(self, ctx, call_next, meta):
        last: Exception | None = None
        for attempt in range(1, meta["max_attempts"] + 1):
            try:
                return await (call_next(ctx) if attempt == 1 else ctx.method(*ctx.args, **ctx.kwargs))
            except meta["retry_on"] as exc:
                last = exc
                self._log(ctx, attempt, meta, exc)
                if attempt < meta["max_attempts"]:
                    await asyncio.sleep(self._delay(meta, attempt))
        raise RetryExhaustedError(ctx.name, meta["max_attempts"], last)

    @staticmethod
    def _delay(meta, attempt: int) -> float:
        return meta["backoff_seconds"] * (2 ** (attempt - 1))

    @staticmethod
    def _log(ctx, attempt, meta, exc):
        logger.warning("%s.%s attempt %d/%d failed: %s", ctx.cls.__name__, ctx.name, attempt, meta["max_attempts"], exc)

CircuitBreakerInterceptor

Per-method circuit: closed -> open -> half-open -> closed.

Source code in src/pico_resilience/interceptors.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@component(scope="singleton")
class CircuitBreakerInterceptor:
    """Per-method circuit: closed -> open -> half-open -> closed."""

    def __init__(self, settings: ResilienceSettings):
        self.settings = settings
        self._states: Dict[str, _CircuitState] = {}
        # ponytail: one global lock; per-circuit locks if contention ever matters.
        self._lock = threading.Lock()

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        meta = _meta(ctx, CIRCUIT_META)
        if not self.settings.enabled or not meta:
            return call_next(ctx)
        key = f"{ctx.cls.__module__}.{ctx.cls.__qualname__}.{ctx.name}"
        if _is_async(ctx):
            return self._run_async(ctx, call_next, meta, key)
        self._check(key, meta, ctx)
        try:
            result = call_next(ctx)
        except Exception:
            self._record_failure(key, meta, ctx)
            raise
        self._record_success(key)
        return result

    async def _run_async(self, ctx, call_next, meta, key):
        self._check(key, meta, ctx)
        try:
            result = await call_next(ctx)
        except Exception:
            self._record_failure(key, meta, ctx)
            raise
        self._record_success(key)
        return result

    def _check(self, key: str, meta, ctx) -> None:
        with self._lock:
            st = self._states.setdefault(key, _CircuitState())
            if st.opened_at is None:
                return
            elapsed = time.monotonic() - st.opened_at
            if elapsed < meta["reset_timeout_seconds"]:
                raise CircuitOpenError(ctx.name, meta["reset_timeout_seconds"] - elapsed)
            # half-open: this call is the trial; one failure reopens
            st.opened_at = None
            st.failures = meta["failure_threshold"] - 1

    def _record_failure(self, key: str, meta, ctx) -> None:
        with self._lock:
            st = self._states.setdefault(key, _CircuitState())
            st.failures += 1
            if st.failures >= meta["failure_threshold"]:
                st.opened_at = time.monotonic()
                logger.error("circuit OPEN for %s.%s after %d failures", ctx.cls.__name__, ctx.name, st.failures)

    def _record_success(self, key: str) -> None:
        with self._lock:
            st = self._states.setdefault(key, _CircuitState())
            st.failures = 0
            st.opened_at = None

TimeoutInterceptor

Bounds async methods with asyncio.timeout.

Source code in src/pico_resilience/interceptors.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@component(scope="singleton")
class TimeoutInterceptor:
    """Bounds async methods with ``asyncio.timeout``."""

    def __init__(self, settings: ResilienceSettings):
        self.settings = settings

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        meta = _meta(ctx, TIMEOUT_META)
        if not self.settings.enabled or not meta:
            return call_next(ctx)
        return self._bounded(ctx, call_next, meta["seconds"])

    @staticmethod
    async def _bounded(ctx, call_next, seconds: float):
        try:
            async with asyncio.timeout(seconds):
                return await call_next(ctx)
        except TimeoutError:
            raise TimeoutError(f"{ctx.cls.__name__}.{ctx.name}() exceeded {seconds}s") from None

pico_resilience.config

Settings for the resilience interceptors (prefix resilience, zero-config).

ResilienceSettings dataclass

enabled: false turns every interceptor into a pass-through.

Source code in src/pico_resilience/config.py 
 8
 9
10
11
12
13
@configured(target="self", prefix="resilience", mapping="tree")
@dataclass
class ResilienceSettings:
    """``enabled: false`` turns every interceptor into a pass-through."""

    enabled: bool = True

pico_resilience.exceptions

Exceptions for pico-resilience.

PicoResilienceError

Bases: Exception

Base exception for all pico-resilience errors.

Source code in src/pico_resilience/exceptions.py 
4
5
class PicoResilienceError(Exception):
    """Base exception for all pico-resilience errors."""

RetryExhaustedError

Bases: PicoResilienceError

Raised when a @retryable method fails on every attempt.

Attributes:

Name Type Description
method_name

The method that kept failing.
attempts

How many attempts were made.
last_error

The exception raised by the final attempt.
Source code in src/pico_resilience/exceptions.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class RetryExhaustedError(PicoResilienceError):
    """Raised when a ``@retryable`` method fails on every attempt.

    Attributes:
        method_name: The method that kept failing.
        attempts: How many attempts were made.
        last_error: The exception raised by the final attempt.
    """

    def __init__(self, method_name: str, attempts: int, last_error: Exception):
        self.method_name = method_name
        self.attempts = attempts
        self.last_error = last_error
        super().__init__(f"{method_name}() failed after {attempts} attempts: {last_error}")

CircuitOpenError

Bases: PicoResilienceError

Raised when a call is rejected because the circuit is open.

Attributes:

Name Type Description
method_name

The protected method.
retry_after_seconds

Seconds until the circuit allows a trial call.
Source code in src/pico_resilience/exceptions.py 
24
25
26
27
28
29
30
31
32
33
34
35
class CircuitOpenError(PicoResilienceError):
    """Raised when a call is rejected because the circuit is open.

    Attributes:
        method_name: The protected method.
        retry_after_seconds: Seconds until the circuit allows a trial call.
    """

    def __init__(self, method_name: str, retry_after_seconds: float):
        self.method_name = method_name
        self.retry_after_seconds = retry_after_seconds
        super().__init__(f"circuit for {method_name}() is open; retry in {retry_after_seconds:.1f}s")