tenacity

repository·main·Indexed 26 days ago

https://github.com/jd/tenacity

A general-purpose retrying library for Python designed to simplify adding retry behavior to functions, coroutines, and code blocks. It provides a generic decorator API (@retry), class-based control (Retrying, AsyncRetrying), and customizable stop, wait, and retry conditions. Tenacity supports asyncio, Trio, and Tornado, and allows for detailed configuration of exponential backoff, jitter, and lifecycle hooks via callbacks.

Tokens
6.4K
Snippets
13
Records
55
Agent score
92%

What's inside tenacity

  1. Overview of tenacity features

    main

    Tenacity is a general-purpose retrying library for Python that provides several ways to control retry behavior:

    • Generic Decorator API: Easily add retry logic to functions.
    • Stop conditions: Limit retries by the number of attempts.
    • Wait conditions: Control the delay between attempts (e.g., exponential backoff).
    • Exception customization: Specify which exceptions trigger a retry.
    • Result customization: Retry based on the value returned by the function.
    • Coroutine support: Retry asynchronous functions.
    • Context manager: Retry a specific block of code using a context manager.
  2. Configure wait intervals between retries

    main

    Control the delay between retry attempts using the wait parameter. Supported patterns include:

    • wait_fixed(s): A constant delay of s seconds.
    • wait_random(min=m, max=M): A random delay between m and M seconds.
    • wait_exponential(multiplier=n, min=m, max=M): Exponential backoff using $n \times 2^{attempt}$ seconds, bounded by min and max.
    • wait_fixed(s) + wait_random(m, M): Combines a fixed delay with jitter.
    • wait_random_exponential(multiplier=n, max=M): Exponential backoff with random jitter.
    • wait_chain(*waiters): A sequence of different wait strategies applied in order.
    @retry(wait=wait_fixed(2))
    def wait_2_s():
        raise Exception
    
    @retry(wait=wait_random(min=1, max=2))
    def wait_random_1_to_2_s():
        raise Exception
    
    @retry(wait=wait_exponential(multiplier=1, min=4, max=10))
    def wait_exponential_1():
        raise Exception
    
    @retry(wait=wait_fixed(3) + wait_random(0, 2))
    def wait_fixed_jitter():
        raise Exception
    
    @retry(wait=wait_random_exponential(multiplier=1, max=60))
    def wait_exponential_jitter():
        raise Exception
    
    @retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] + [wait_fixed(7) for i in range(2)] + [wait_fixed(9)]))
    def wait_fixed_chained():
        raise Exception
  3. Define retry conditions based on exceptions or results

    main

    Use the retry parameter to specify which failures should trigger a retry. You can filter by exception type, exception message, or the function's return value.

    Exception-based retries:

    • retry_if_exception_type(ExceptionClass): Retry if a specific exception type is raised.
    • retry_if_not_exception_type(ExceptionClass): Retry if any exception other than the specified type is raised.
    • retry_if_exception(predicate): Use a custom predicate function.
    • retry_if_exception_message(message_predicate): Retry based on the exception message.

    Result-based retries:

    • retry_if_result(predicate): Retry if the function's return value satisfies the predicate.
    • retry_if_not_result(predicate): Retry if the return value does not satisfy the predicate.

    Combinators:

    • retry_any: Retry if any of the provided conditions are met.
    • retry_all: Retry only if all provided conditions are met.
    • Use the | operator to combine conditions (e.g., retry_if_result(is_none) | retry_if_exception_type(IOError)).
    class ClientError(Exception): pass
    
    @retry(retry=retry_if_exception_type(IOError))
    def might_io_error():
        raise Exception
    
    @retry(retry=retry_if_not_exception_type(ClientError))
    def might_client_error():
        raise ClientError
    
    def is_none_p(value): return value is None
    
    @retry(retry=retry_if_result(is_none_p))
    def might_return_none():
        return None
    
    @retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type(IOError)))
    def combined_condition():
        pass
  4. Use callbacks for logging and lifecycle hooks

    main

    Tenacity provides hooks to execute code at different stages of the retry lifecycle. These callbacks accept a retry_state object.

    • before: Executed before every attempt.
    • after: Executed after every attempt (even successful ones).
    • before_sleep: Executed after a failure, just before the wait interval begins. Ideal for logging or re-establishing connections (e.g., refreshing tokens).
    • retry_error_callback: Executed when all retries have failed. Can be used to return a fallback value instead of raising an exception.

    Built-in logging helpers:

    • before_log(logger, level)
    • after_log(logger, level)
    • before_sleep_log(logger, level)
    import logging
    import sys
    
    logger = logging.getLogger(__name__)
    
    # Using built-in logging helpers
    @retry(stop=stop_after_attempt(3), before_sleep=before_sleep_log(logger, logging.DEBUG))
    def raise_my_exception():
        raise MyException("Fail")
    
    # Using a custom callback to return a fallback value
    def return_last_value(retry_state):
        return retry_state.outcome.result()
    
    @retry(stop=stop_after_attempt(3), 
           retry_error_callback=return_last_value, 
           retry=retry_if_result(lambda x: x is False))
    def eventually_return_false():
        return False
  5. Handle retry errors and reraise exceptions

    main

    By default, when all retry attempts are exhausted, Tenacity raises a RetryError. The original exception encountered is nested within the RetryError stack trace.

    To make the original exception appear at the end of the stack trace (making it easier to debug), set reraise=True in the @retry decorator.

    @retry(reraise=True, stop=stop_after_attempt(3))
    def raise_my_exception():
        raise MyException("Fail")
    
    try:
        raise_my_exception()
    except MyException:
        # The original MyException is caught here because reraise=True
        pass
  6. Set retry stop conditions

    main

    You can define when a retry loop should stop using stop parameters. Common strategies include stopping after a specific number of attempts or after a total elapsed time.

    • stop_after_attempt(n): Stops after n attempts.
    • stop_after_delay(s): Stops after s seconds have elapsed.
    • stop_before_delay(s): Stops one attempt before the delay s would be exceeded.
    • Combining conditions: Use the | operator to combine multiple stop conditions (e.g., stop after 10 seconds OR 5 attempts).
    @retry(stop=stop_after_attempt(7))
    def stop_after_7_attempts():
        raise Exception
    
    @retry(stop=stop_after_delay(10))
    def stop_after_10_s():
        raise Exception
    
    @retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
    def stop_after_10_s_or_5_retries():
        raise Exception
  7. Disable retries

    main

    You can disable retries globally via an environment variable or on a per-call basis using retry_with.

    • Environment Variable: Use enabled=os.getenv(...) != '0' in the decorator.
    • Per-call: Use func.retry_with(enabled=False)(). This is useful for speeding up tests.
    import os
    
    @retry(
        enabled=os.getenv("ENABLE_RETRIES", "1") != "0",
        stop=stop_after_attempt(5),
        wait=wait_fixed(1),
    )
    def call_api():
        pass
    
    # Disable for a specific call (e.g. in tests)
    call_api.retry_with(enabled=False)()
  8. Configure retry behavior using keyword arguments

    main

    The tenacity.retry decorator and Retrying classes allow you to customize retry behavior using several specialized keyword arguments. You can pass functions or objects from the following modules to these arguments:

    • retry: Use functions from tenacity.retry to define the condition for retrying (e.g., retrying on specific exceptions).
    • stop: Use functions from tenacity.stop to define when retrying should cease (e.g., after a certain number of attempts or a timeout).
    • wait: Use functions from tenacity.wait to define how long to wait between retry attempts (e.g., fixed delay, exponential backoff).
    • before: Use functions from tenacity.before to execute code before every retry attempt.
    • after: Use functions from tenacity.after to execute code after every retry attempt.
    • before_sleep: Use functions from tenacity.before_sleep to execute code specifically before the thread/task sleeps between attempts.
    • sleep: Use functions from tenacity.nap to define a custom sleep mechanism.
  9. Retry code blocks without a function

    main

    You can retry a specific block of code within an existing function using the Retrying class as a context manager inside a loop.

    from tenacity import Retrying, RetryError, stop_after_attempt
    
    try:
        for attempt in Retrying(stop=stop_after_attempt(3)):
            with attempt:
                # Your code block here
                raise Exception('My code is failing!')
    except RetryError:
        pass
  10. Retry asynchronous code

    main

    Tenacity supports asyncio, Trio, and Tornado. Use the @retry decorator on async def functions, or use AsyncRetrying for manual loops. Sleeps are handled asynchronously automatically.

    If using a non-standard event loop (like Trio), you may need to pass the specific sleep function to the sleep parameter.

  11. Use the @retry decorator to wrap functions

    main

    The most common way to use Tenacity is by applying the @retry decorator to a function. You can use it without arguments to apply default retry behavior, or with arguments to configure specific stop, wait, and retry strategies.

    Note: When using @retry as a decorator, it supports both @retry and @retry() syntax.