Skip to content

windlass.core.retry

retry

Retry with exponential backoff and jitter.

Provider calls fail transiently: rate limits, 5xx, socket resets. Windlass wraps every outbound call in :func:retry_async so adapters do not each reinvent a backoff loop, and so the policy is configurable in one place (:class:~windlass.core.config.RetryConfig).

Only transient failures are retried. A 401 or a schema validation error is raised immediately — retrying those just wastes a minute before failing anyway.

is_retryable 

is_retryable(exc: BaseException, extra: Iterable[type[BaseException]] = ()) -> bool

Decide whether exc is worth another attempt.

The decision order matters: an explicit non-retryable class wins over a retryable base class, so :class:AuthenticationError never retries even though it subclasses ProviderError.

Parameters:

Name Type Description Default
exc BaseException

The raised exception.

 required
extra Iterable[type[BaseException]]

Additional exception types to treat as retryable — adapters pass their SDK's transient error classes here.

 ()

Returns:

Type Description
bool

True when a retry is warranted.
Example

is_retryable(RateLimitError("slow down")) True is_retryable(AuthenticationError("bad key")) False

Source code in src\windlass\core\retry.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def is_retryable(exc: BaseException, extra: Iterable[type[BaseException]] = ()) -> bool:
    """Decide whether ``exc`` is worth another attempt.

    The decision order matters: an explicit non-retryable class wins over a
    retryable base class, so :class:`AuthenticationError` never retries even
    though it subclasses ``ProviderError``.

    Args:
        exc: The raised exception.
        extra: Additional exception types to treat as retryable — adapters pass
            their SDK's transient error classes here.

    Returns:
        True when a retry is warranted.

    Example:
        >>> is_retryable(RateLimitError("slow down"))
        True
        >>> is_retryable(AuthenticationError("bad key"))
        False
    """
    if isinstance(exc, NON_RETRYABLE):
        return False
    if isinstance(exc, RETRYABLE) or (tuple(extra) and isinstance(exc, tuple(extra))):
        return True
    # Duck-typed HTTP status: httpx / SDK errors expose .status_code or .response.
    status = getattr(exc, "status_code", None)
    if status is None:
        status = getattr(getattr(exc, "response", None), "status_code", None)
    if isinstance(status, int):
        return status == 408 or status == 429 or 500 <= status < 600
    return False

backoff_delays 

backoff_delays(config: RetryConfig | None = None) -> list[float]

Return the delay schedule a retry loop would use.

Useful for tests and for logging the policy at startup.

Parameters:

Name Type Description Default
config RetryConfig | None

Policy to expand. Defaults to the configured global policy.

 None

Returns:

Type Description
list[float]

One delay per retry (so attempts - 1 entries), jitter excluded.
Example

backoff_delays(RetryConfig(attempts=4, initial_delay=1, multiplier=2)) [1.0, 2.0, 4.0]

Source code in src\windlass\core\retry.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def backoff_delays(config: RetryConfig | None = None) -> list[float]:
    """Return the delay schedule a retry loop would use.

    Useful for tests and for logging the policy at startup.

    Args:
        config: Policy to expand. Defaults to the configured global policy.

    Returns:
        One delay per retry (so ``attempts - 1`` entries), jitter excluded.

    Example:
        >>> backoff_delays(RetryConfig(attempts=4, initial_delay=1, multiplier=2))
        [1.0, 2.0, 4.0]
    """
    cfg = config or settings().retry
    delays: list[float] = []
    delay = cfg.initial_delay
    for _ in range(max(0, cfg.attempts - 1)):
        delays.append(min(delay, cfg.max_delay))
        delay *= cfg.multiplier
    return delays

retry_async async 

retry_async(
    fn: Callable[[], Awaitable[_T]],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    description: str = "provider call"
) -> _T

Await fn with retries on transient failures.

Parameters:

Name Type Description Default
fn Callable[[], Awaitable[_T]]

A zero-argument callable returning an awaitable. It is re-invoked on each attempt, so pass a lambda rather than a coroutine object — coroutines cannot be awaited twice.

 required
config RetryConfig | None

Retry policy. Defaults to the global configuration.

 None
retry_on Iterable[type[BaseException]]

Extra exception types to treat as transient.

 ()
on_retry Callable[[int, BaseException, float], None] | None

Callback invoked as (attempt, exception, delay) before each sleep. Handy for metrics.

 None
description str

Label used in log messages.

 'provider call'

Returns:

Type Description
_T

Whatever fn eventually returns.

Raises:

Type Description
Exception

The last failure, once attempts are exhausted or the error is classified as non-retryable.
Example

import asyncio calls = [] async def flaky(): ... calls.append(1) ... if len(calls) < 2: ... raise ConnectionError("reset") ... return "ok" asyncio.run(retry_async(flaky, config=RetryConfig(attempts=3, initial_delay=0))) 'ok'

Source code in src\windlass\core\retry.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
async def retry_async(
    fn: Callable[[], Awaitable[_T]],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    description: str = "provider call",
) -> _T:
    """Await ``fn`` with retries on transient failures.

    Args:
        fn: A zero-argument callable returning an awaitable. It is re-invoked on
            each attempt, so pass a lambda rather than a coroutine object —
            coroutines cannot be awaited twice.
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra exception types to treat as transient.
        on_retry: Callback invoked as ``(attempt, exception, delay)`` before each
            sleep. Handy for metrics.
        description: Label used in log messages.

    Returns:
        Whatever ``fn`` eventually returns.

    Raises:
        Exception: The last failure, once attempts are exhausted or the error is
            classified as non-retryable.

    Example:
        >>> import asyncio
        >>> calls = []
        >>> async def flaky():
        ...     calls.append(1)
        ...     if len(calls) < 2:
        ...         raise ConnectionError("reset")
        ...     return "ok"
        >>> asyncio.run(retry_async(flaky, config=RetryConfig(attempts=3, initial_delay=0)))
        'ok'
    """
    cfg = config or settings().retry
    last: BaseException | None = None

    for attempt in range(1, cfg.attempts + 1):
        try:
            return await fn()
        except BaseException as exc:
            last = exc
            if attempt >= cfg.attempts or not is_retryable(exc, retry_on):
                raise
            delay = _next_delay(attempt, cfg, exc)
            _log.warning(
                "%s failed (attempt %d/%d): %s — retrying in %.2fs",
                description,
                attempt,
                cfg.attempts,
                exc,
                delay,
            )
            if on_retry is not None:
                on_retry(attempt, exc, delay)
            await asyncio.sleep(delay)

    raise last if last is not None else WindlassError(f"{description} failed without an exception")

retry_sync 

retry_sync(
    fn: Callable[[], _T],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    description: str = "provider call"
) -> _T

Blocking counterpart of :func:retry_async.

Parameters:

Name Type Description Default
fn Callable[[], _T]

Zero-argument callable, re-invoked per attempt.

 required
config RetryConfig | None

Retry policy. Defaults to the global configuration.

 None
retry_on Iterable[type[BaseException]]

Extra transient exception types.

 ()
description str

Label used in log messages.

 'provider call'

Returns:

Type Description
_T

Whatever fn eventually returns.

Raises:

Type Description
Exception

The last failure once attempts are exhausted.
Example

retry_sync(lambda: 7, config=RetryConfig(attempts=1)) 7

Source code in src\windlass\core\retry.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def retry_sync(
    fn: Callable[[], _T],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    description: str = "provider call",
) -> _T:
    """Blocking counterpart of :func:`retry_async`.

    Args:
        fn: Zero-argument callable, re-invoked per attempt.
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra transient exception types.
        description: Label used in log messages.

    Returns:
        Whatever ``fn`` eventually returns.

    Raises:
        Exception: The last failure once attempts are exhausted.

    Example:
        >>> retry_sync(lambda: 7, config=RetryConfig(attempts=1))
        7
    """
    cfg = config or settings().retry
    last: BaseException | None = None

    for attempt in range(1, cfg.attempts + 1):
        try:
            return fn()
        except BaseException as exc:
            last = exc
            if attempt >= cfg.attempts or not is_retryable(exc, retry_on):
                raise
            delay = _next_delay(attempt, cfg, exc)
            _log.warning(
                "%s failed (attempt %d/%d): %s — retrying in %.2fs",
                description,
                attempt,
                cfg.attempts,
                exc,
                delay,
            )
            time.sleep(delay)

    raise last if last is not None else WindlassError(f"{description} failed without an exception")

with_retry 

with_retry(
    *, config: RetryConfig | None = None, retry_on: Iterable[type[BaseException]] = ()
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator applying the retry policy to a sync or async function.

Parameters:

Name Type Description Default
config RetryConfig | None

Retry policy. Defaults to the global configuration.

 None
retry_on Iterable[type[BaseException]]

Extra transient exception types.

 ()

Returns:

Type Description
Callable[[Callable[..., Any]], Callable[..., Any]]

A decorator preserving the wrapped function's signature and sync/async
Callable[[Callable[..., Any]], Callable[..., Any]]

nature.
Example

@with_retry(config=RetryConfig(attempts=2, initial_delay=0)) ... def ping() -> str: ... return "pong" ping() 'pong'

Source code in src\windlass\core\retry.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def with_retry(
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator applying the retry policy to a sync or async function.

    Args:
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra transient exception types.

    Returns:
        A decorator preserving the wrapped function's signature and sync/async
        nature.

    Example:
        >>> @with_retry(config=RetryConfig(attempts=2, initial_delay=0))
        ... def ping() -> str:
        ...     return "pong"
        >>> ping()
        'pong'
    """

    def decorate(fn: Callable[..., Any]) -> Callable[..., Any]:
        label = getattr(fn, "__qualname__", repr(fn))

        if asyncio.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                return await retry_async(
                    lambda: fn(*args, **kwargs),
                    config=config,
                    retry_on=retry_on,
                    description=label,
                )

            return async_wrapper

        @functools.wraps(fn)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            return retry_sync(
                lambda: fn(*args, **kwargs),
                config=config,
                retry_on=retry_on,
                description=label,
            )

        return sync_wrapper

    return decorate