Python Tenacity: Configure Retries, Stops, and Waits
This article provides a practical overview of how to build resilient Python applications using the Tenacity library. It details the configuration of stop conditions to define when retries should terminate, wait strategies to control the delay between attempts, and retry conditions to specify which exceptions or results trigger a retry.
The @retry Decorator
Tenacity operates primarily through the @retry decorator
applied to functions. By default, @retry will retry an
operation indefinitely whenever an unhandled exception is raised, with
no delay between calls. To control this behavior, you pass arguments
such as stop, wait, and retry
into the decorator.
from tenacity import retry
@retry
def basic_task():
# Retries infinitely on any exception without waiting
passConfiguring Stop Conditions
Stop strategies determine when Tenacity ceases retrying and re-raises the caught exception.
1. stop_after_attempt
Stops execution after a designated number of attempts.
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def call_api():
# Will attempt at most 3 times
pass2. stop_after_delay
Stops retrying once a specific elapsed time (in seconds) has passed since the initial invocation.
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(10))
def query_database():
# Stops retrying after 10 seconds total duration
pass3. Combining Stops
Stop conditions can be combined using the bitwise OR (|)
operator.
from tenacity import retry, stop_after_attempt, stop_after_delay
@retry(stop=(stop_after_attempt(5) | stop_after_delay(20)))
def upload_file():
# Stops after 5 attempts OR 20 seconds, whichever comes first
passConfiguring Wait Intervals
Wait strategies define the duration the system pauses between consecutive retry attempts.
1. wait_fixed
Pauses for a constant duration between attempts.
from tenacity import retry, wait_fixed
@retry(wait=wait_fixed(2))
def fetch_data():
# Waits 2 seconds between each attempt
pass2. wait_random
Introduces jitter by pausing for a random duration within a defined range.
from tenacity import retry, wait_random
@retry(wait=wait_random(min=1, max=3))
def ping_service():
# Waits between 1 and 3 seconds
pass3. wait_exponential
Implements exponential backoff, progressively multiplying the wait interval.
from tenacity import retry, wait_exponential
@retry(wait=wait_exponential(multiplier=1, min=2, max=10))
def process_queue():
# Wait starts at 2s, doubles each try, capped at 10s
pass4. Combining and Chaining Waits
Waits can be added together (+) to introduce jitter to a
fixed duration, or sequenced using wait_chain.
from tenacity import retry, wait_fixed, wait_random, wait_chain
# Fixed wait with random jitter
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def sync_data():
pass
# Explicit intervals per attempt
@retry(wait=wait_chain(*[wait_fixed(1) for _ in range(2)] + [wait_fixed(5)]))
def staged_wait():
# Waits 1s for the first two retries, then 5s for subsequent retries
passConfiguring Retry Conditions
Retry conditions specify which triggers—exceptions or return values—should initiate a retry rather than failing immediately or accepting the result.
1. Exception Filtering
By default, any exception triggers a retry. You can restrict this to
specific exceptions using retry_if_exception_type.
import requests
from tenacity import retry, retry_if_exception_type
@retry(retry=retry_if_exception_type(requests.exceptions.Timeout))
def send_request():
# Only retries on Timeout errors; any other exception raises immediately
pass2. Result Inspection
You can trigger retries based on the returned value using
retry_if_result.
from tenacity import retry, retry_if_result
def is_empty_response(result):
return result is None or len(result) == 0
@retry(retry=retry_if_result(is_empty_response))
def poll_job():
# Retries if the returned value is None or empty
return []3. Combining Retry Conditions
Conditions can be combined using bitwise operators (|
for OR, & for AND).
from tenacity import retry, retry_if_exception_type, retry_if_result
@retry(retry=(retry_if_exception_type(IOError) | retry_if_result(lambda res: res is False)))
def write_stream():
# Retries on IOError OR if the function returns False
passUnified Example
Below is a production-ready configuration combining stop parameters, exponential backoff with jitter, and custom exception filtering:
import requests
from tenacity import (
retry,
stop_after_attempt,
stop_after_delay,
wait_exponential,
wait_random,
retry_if_exception_type
)
@retry(
stop=(stop_after_attempt(5) | stop_after_delay(30)),
wait=wait_exponential(multiplier=1, min=1, max=10) + wait_random(0, 1),
retry=retry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)),
reraise=True
)
def execute_resilient_request(url: str):
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()