DOCSEARCHbyAlgolia
Submit Docs

stamina

repository·main·Indexed 23 days ago

https://github.com/hynek/stamina

A production-grade retry library providing an ergonomic, opinionated wrapper around Tenacity. It features sensible defaults including exponential backoff with jitter, limits on attempts and total time, and native support for synchronous and asynchronous callables (including Trio). It includes built-in instrumentation for Prometheus, structlog, and standard library logging, as well as dedicated testing modes to deactivate retries or remove backoff delays.

Tokens
7K
Snippets
12
Records
46
Agent score
31%

What's inside stamina

  1. Overview of stamina features

    main

    stamina is an opinionated wrapper around Tenacity designed for ergonomics and safety in distributed systems. Key features include:

    • Selective Retries: Retry only on specific exceptions or use a backoff hook to introspect exceptions.
    • Resilient Backoff: Exponential backoff with jitter is used by default.
    • Safety Limits: Limits both the number of retries (attempts) and the total elapsed time.
    • Async Support: Native support for async callables and Trio.
    • Type Safety: Preserves type hints of the decorated callable.
    • Instrumentation: Out-of-the-box support for Prometheus, structlog, and standard library logging.
    • Testability: Dedicated support for testing that allows globally deactivating retries, limiting retry counts, or removing backoffs.

  2. Understand the default retry strategy in stamina

    main

    By default, stamina.retry implements an exponential backoff with jitter to prevent cascading failures and thundering herd problems.

    Default Behavior:

    • Initial Backoff: Starts at 100ms.
    • Growth: Increases exponentially by a factor of 2.
    • Cap: The backoff increases until it reaches 5 seconds, where it stays.
    • Maximum Duration/Attempts: The retry cycle continues until 45 seconds have passed or 10 attempts have been made.
    • Jitter: A random jitter between 0 and 1.0 seconds is added at every step until the 5-second maximum is reached. No additional jitter is added once the 5-second cap is hit.

    Mathematical Formula: min(5.0, 0.1 * 2^{attempt - 1} + random(0, 1.0))

    This means the first backoff is at most 1.1 seconds, and subsequent backoffs are capped at 5 seconds.

  3. How instrumentation hooks work in stamina

    main

    Stamina provides instrumentation hooks that are triggered whenever a retry is scheduled, but before the backoff wait begins. This allows you to monitor failures and scheduled retries immediately.

    Hook Mechanics

    • Input: A hook is a callable that accepts a single stamina.instrumentation.RetryDetails object.
    • Return Value: The return value is ignored unless it is a context manager.
    • Context Managers: If a hook returns a context manager, it is entered when the retry is scheduled and exited right before the retry is attempted. This is useful for measuring the duration of the sleep/backoff period (e.g., for emitting spans).

    Lazy Initialization

    For environments like CLI tools where you might want to delay instrumentation setup until the first retry occurs, you can use stamina.instrumentation.RetryHookFactory. You pass a callable that creates and returns a retry hook to stamina.instrumentation.set_on_retry_hooks wrapped in the factory.

  4. Use backoff hooks for fine-grained retry logic

    main

    If a simple exception type is too broad (e.g., you want to retry on 5xx errors but not 404s), you can pass a backoff hook to the on parameter.

    A backoff hook is a callable that accepts the raised exception and returns:

    1. A bool: True to retry, False to stop.
    2. A float (seconds) or datetime.timedelta: Specifies a custom backoff delay, overriding the default exponential backoff machinery. This is useful for respecting Retry-After headers.
    def retry_only_on_real_errors(exc: Exception) -> bool:
    # If the error is an HTTP status error, only retry on 5xx errors.
    if isinstance(exc, httpx2.HTTPStatusError):
        return exc.response.status_code >= 500

    # Otherwise retry on all httpx2 errors.
    return isinstance(exc, httpx2.HTTPError)

@stamina.retry(on=retry_only_on_real_errors, attempts=3)
def do_it(code: int) -> httpx2.Response:
    resp = httpx2.get(f"https://httpbin.org/status/{code}")
    resp.raise_for_status()
    return resp

  5. Observability and Logging

    main

    Stamina provides built-in observability:

    • Metrics: If prometheus-client is installed, it increments the stamina_retries_total counter.
    • Logs: It uses structlog for logging retries, falling back to the standard logging module if structlog is not available.

  6. Disable retries globally for testing

    main

    If you are using the decorator-based API and want to prevent any retry logic from executing during your test suite, you can use stamina.set_active(False). This is most effectively used within a pytest fixture with autouse=True and scope="session" to ensure retries are disabled across all tests.

    import pytest
import stamina


@pytest.fixture(autouse=True, scope="session")
def deactivate_retries():
    stamina.set_active(False)

  7. Tune retry parameters in stamina.retry

    main

    While the default exponential backoff with jitter is a robust starting point for most distributed systems, you can tune the parameters to suit your specific needs (e.g., different backoff durations for network hiccups vs. overloaded databases).

    Refer to the stamina.retry API documentation to customize the backoff timing, jitter, and maximum attempt/duration limits.

  8. Use testing mode to limit backoff and attempts

    main

    When using iterator-based APIs like stamina.retry_context, you may want to test the retry logic itself without waiting for exponential backoff or performing hundreds of attempts.

    Calling stamina.set_testing(True) enables a dedicated testing mode that:

    1. Disables all backoff (retries happen immediately).
    2. Caps the number of attempts to a single attempt by default.

    You can override the attempt cap by passing the attempts argument to stamina.set_testing(True, attempts=N). Always remember to call stamina.set_testing(False) to restore normal behavior after your tests.

    import stamina

stamina.set_testing(True)  # no backoff, 1 attempt
stamina.set_testing(True, attempts=2)  # no backoff, 2 attempts

for attempt in stamina.retry_context(on=ValueError, attempts=1_000):
    with attempt:
        print("trying", attempt.num)
        raise ValueError("nope")

stamina.set_testing(False)  # back to business as usual

  9. Use stamina with async/await (asyncio and Trio)

    main

    Stamina supports both asyncio and Trio. Use the @stamina.retry decorator on async def functions or use async for with stamina.retry_context().

    When configuring retries for async functions, you can pass datetime.timedelta objects to timeout, wait_initial, wait_max, and wait_jitter.

    import datetime as dt

@stamina.retry(
    on=httpx2.HTTPError, attempts=3, timeout=dt.timedelta(seconds=10)
)
async def do_it_async(code: int) -> httpx2.Response:
    async with httpx2.AsyncClient() as client:
        resp = await client.get(f"https://httpbin.org/status/{code}")
    resp.raise_for_status()
    return resp


async def with_block(code: int) -> httpx2.Response:
    async for attempt in stamina.retry_context(
        on=httpx2.HTTPError, attempts=3
    ):
        with attempt:
            async with httpx2.AsyncClient() as client:
                resp = await client.get(f"https://httpbin.org/status/{code}")
            resp.raise_for_status()

    return resp

  10. Implement a RetryHook to monitor retry attempts

    main

    A RetryHook is a callable used to perform actions after a retry has been scheduled. It receives a RetryDetails object containing metadata about the failed attempt.

    Starting from version 25.1.0, a RetryHook can return a context manager. If it does, the context manager is entered when the retry is scheduled and exited immediately before the next retry attempt is made. This is useful for managing resources (like database connections or logs) that need to be reset or re-established specifically for the retry attempt.

  11. Use stamina.retry to retry failed operations

    main

    Use the @stamina.retry decorator to wrap functions or methods that are prone to transient failures. By default, stamina implements production-ready retry logic including exponential backoff with jitter, limits on both the number of attempts and total time, and support for both synchronous and asynchronous callables.

    You can specify which exceptions to retry on using the on parameter.

    import httpx2

import stamina

@stamina.retry(on=httpx2.HTTPError, attempts=3)
def do_it(code: int) -> httpx2.Response:
    resp = httpx2.get(f"https://httpbin.org/status/{code}")
    resp.raise_for_status()

    return resp

  12. Add retries using the @stamina.retry decorator

    main

    The simplest way to implement retries is to decorate a function or method with @stamina.retry(). You must explicitly specify which exceptions to retry on using the on parameter; stamina does not retry on Exception by default to prevent accidental infinite loops or incorrect error handling.

    Common parameters:

    • on: An exception class or a backoff hook (see below).
    • attempts: The maximum number of attempts to make.
    import httpx2
import stamina

@stamina.retry(on=httpx2.HTTPError, attempts=3)
def do_it(code: int) -> httpx2.Response:
    resp = httpx2.get(f"https://httpbin.org/status/{code}")
    resp.raise_for_status()

    return resp