Python Daily Library: Tenacity - A Powerful Retry Library

📚 Introduction

<span>tenacity</span> is one of the most powerful and flexible retry libraries in Python, allowing you to elegantly add an automatic retry mechanism to functions to handle issues such as network request failures, connection timeouts, and temporary database unavailability.

It is the modern successor to <span>retrying</span>, supporting decorator syntax, asynchronous functions, and rich retry control logic.

🌟 Core Features

Feature	Description
Decorator Syntax	Concise and easy-to-use `<span>@retry</span>` decorator
Asynchronous Support	Full support for retrying asynchronous functions
Flexible Strategies	Multiple combinations of stop, wait, and retry strategies
Hook Functions	Support for callback functions before and after retries
Detailed Logging	Built-in detailed logging of the retry process
Type Hints	Full support for type hints

🚀 Installation

pip install tenacity

✅ Basic Usage

1. Simple Retry

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def unstable_func():
    print("Trying...")
    raise Exception("Failed")
unstable_func()

Explanation: stop_after_attempt(3): Retry a maximum of 3 times
wait_fixed(2): Wait 2 seconds between each retry

2. Retry with Return Value

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def fetch_data():
    import random
    result = random.randint(1, 10)
    if result < 5:  # Simulate failure
        raise Exception(f"Random failure: {result}")
    return f"Successfully fetched data: {result}"
try:
    result = fetch_data()
    print(result)
except Exception as e:
    print(f"Final failure: {e}")

🔄 Advanced Usage

1. Time-based Retry

from tenacity import retry, stop_after_delay, wait_fixed
@retry(stop=stop_after_delay(10), wait=wait_fixed(1))  # Try for a maximum of 10 seconds
def timeout_func():
    import time
    time.sleep(3)  # Simulate a time-consuming operation
    raise Exception("Timeout operation")
timeout_func()

2. Retry by Exception Type

from tenacity import retry, retry_if_exception_type, stop_after_attempt
@retry(
    retry=retry_if_exception_type((ValueError, ConnectionError)),
    stop=stop_after_attempt(5))
def might_raise():
    import random
    error_type = random.choice([ValueError, ConnectionError, TypeError])
    raise error_type("Random exception")
try:
    might_raise()
except TypeError:
    print("TypeError will not be retried")

3. Asynchronous Function Retry

from tenacity import AsyncRetrying, stop_after_attempt
import asyncio
async def unstable_async():
    async for attempt in AsyncRetrying(stop=stop_after_attempt(3)):
        with attempt:
            print("Trying async...")
            raise Exception("Error")
# Run asynchronous retry
asyncio.run(unstable_async())

4. Conditional Retry (Based on Return Value)

from tenacity import retry, retry_if_result, stop_after_attempt
@retry(
    retry=retry_if_result(lambda x: x is None or x == ""),
    stop=stop_after_attempt(3))
def fetch_user_data(user_id):
    # Simulate a situation that may return a null value
    import random
    if random.random() < 0.7:
        return None  # Simulate failure
    return f"User data: {user_id}"
result = fetch_user_data(123)
print(f"Final result: {result}")

5. Exponential Backoff Strategy

from tenacity import retry, wait_exponential, stop_after_attempt
@retry(
    wait=wait_exponential(multiplier=1, min=4, max=10),
    stop=stop_after_attempt(5))
def exponential_backoff():
    print("Trying...")
    raise Exception("Needs retry")
exponential_backoff()

6. Combined Strategies

from tenacity import (
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_exponential,
    retry_if_exception_type)
@retry(
    stop=(stop_after_attempt(5) | stop_after_delay(30)),  # Max 5 times or 30 seconds
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(ConnectionError))
def complex_retry():
    print("Complex retry strategy...")
    raise ConnectionError("Connection failed")
complex_retry()

7. Hook Functions

from tenacity import retry, stop_after_attempt, wait_fixed
import logging
logging.basicConfig(level=logging.INFO)
def before_retry(retry_state):
    print(f"Preparing for retry {retry_state.attempt_number}")
def after_retry(retry_state):
    print(f"Retry {retry_state.attempt_number} completed")
@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    before=before_retry,
    after=after_retry)
def hook_example():
    print("Executing...")
    raise Exception("Failed")
hook_example()

8. Custom Retry Conditions

from tenacity import retry, retry_if_exception, stop_after_attempt
def is_retryable_exception(exception):
    """Custom retry condition"""
    retryable_errors = [
        "Network timeout",
        "Connection failed",
        "Service temporarily unavailable"
    ]
    return any(error in str(exception) for error in retryable_errors)
@retry(
    retry=retry_if_exception(is_retryable_exception),
    stop=stop_after_attempt(3))
def custom_retry():
    import random
    errors = ["Network timeout", "Connection failed", "Service temporarily unavailable", "Insufficient permissions"]
    raise Exception(random.choice(errors))
try:
    custom_retry()
except Exception as e:
    print(f"Final failure: {e}")

🛠️ Configuration Parameters Explained

Parameter	Meaning	Example
`<span>stop</span>`	Stop retry condition	`<span>stop_after_attempt(3)</span>`
`<span>wait</span>`	Retry wait strategy	`<span>wait_fixed(2)</span>`
`<span>retry</span>`	Retry trigger condition	`<span>retry_if_exception_type(ValueError)</span>`
`<span>before</span>`	Retry before hook	`<span>before_retry</span>`
`<span>after</span>`	Retry after hook	`<span>after_retry</span>`
`<span>reraise</span>`	Whether to raise exceptions	`<span>True/False</span>`
`<span>retry_error_cls</span>`	Custom retry exception class	`<span>CustomRetryError</span>`

Common Stop Strategies

from tenacity import (
    stop_after_attempt,      # Maximum attempts
    stop_after_delay,        # Maximum delay time
    stop_all,               # Combine multiple stop conditions
    stop_any                # Stop if any condition is met)
# Combined stop conditions
stop_condition = stop_all(
    stop_after_attempt(5),
    stop_after_delay(30))

Common Wait Strategies

from tenacity import (
    wait_fixed,             # Fixed wait time
    wait_random,            # Random wait time
    wait_exponential,       # Exponential backoff
    wait_chain,            # Chained wait strategy
    wait_combine           # Combined wait strategies)
# Exponential backoff example
wait_strategy = wait_exponential(
    multiplier=1,           # Multiplier
    min=4,                  # Minimum wait time
    max=10                  # Maximum wait time)

🎯 Practical Application Scenarios

1. HTTP Request Retry

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)))
def fetch_api_data(url):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()
# Usage example
try:
    data = fetch_api_data("https://api.example.com/data")
    print(f"Data fetched successfully: {data}")
except Exception as e:
    print(f"Final failure: {e}")

2. Database Connection Retry

import sqlite3
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
@retry(
    stop=stop_after_attempt(5),
    wait=wait_fixed(2),
    retry=retry_if_exception_type(sqlite3.OperationalError))
def connect_database(db_path):
    return sqlite3.connect(db_path)
# Usage example
try:
    conn = connect_database("app.db")
    print("Database connection successful")
    conn.close()
except Exception as e:
    print(f"Database connection failed: {e}")

3. Asynchronous API Calls

import aiohttp
import asyncio
from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential
async def fetch_async_data(url):
    async for attempt in AsyncRetrying(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=8)
    ):
        with attempt:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    response.raise_for_status()
                    return await response.json()
# Usage example
async def main():
    try:
        data = await fetch_async_data("https://api.example.com/data")
        print(f"Asynchronously fetched data successfully: {data}")
    except Exception as e:
        print(f"Asynchronous fetch failed: {e}")
asyncio.run(main())

4. File Operation Retry

import os
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    retry=retry_if_exception_type((OSError, IOError)))
def read_file_safely(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        return f.read()
# Usage example
try:
    content = read_file_safely("config.txt")
    print(f"File read successfully: {content[:50]}...")
except Exception as e:
    print(f"File read failed: {e}")

🔄 Comparison with Other Retry Libraries

Feature	Tenacity	Retrying	Backoff
Decorator Syntax	✅ Supported	✅ Supported	✅ Supported
Asynchronous Support	✅ Full	❌ Not supported	🟡 Limited
Strategy Combinations	✅ Powerful	🟡 Basic	✅ Supported
Hook Functions	✅ Supported	❌ Not supported	❌ Not supported
Type Hints	✅ Full	❌ Not supported	🟡 Partial
Active Maintenance	✅ Active	❌ Stopped	🟡 Average

⚠️ Cautions

1. Idempotency

# ✅ Ensure the function is idempotent
@retry(stop=stop_after_attempt(3))
def idempotent_function(user_id):
    # This function can be safely retried
    return get_user_data(user_id)
# ❌ Avoid non-idempotent operations
@retry(stop=stop_after_attempt(3))
def non_idempotent_function():
    # This function may cause duplicate operations on retry
    return create_user()

2. Exception Handling

from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def careful_retry():
    try:
        # Potentially failing operation
        result = risky_operation()
        return result
    except ValueError as e:
        # Non-retryable exception
        raise e
    except Exception as e:
        # Retryable exception
        raise e

3. Performance Considerations

# Avoid overly frequent retries
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=1, max=60)  # Reasonable wait time)
def performance_aware_retry():
    # Operation logic
    pass

❓ Frequently Asked Questions (FAQ)

Q1: How to get the retry count?

from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def retry_with_count():
    # You can get the retry count through logging or hook functions
    print("Executing...")
    raise Exception("Failed")

Q2: How to customize retry exceptions?

from tenacity import retry, stop_after_attempt
class CustomRetryError(Exception):
    pass
@retry(
    stop=stop_after_attempt(3),
    retry_error_cls=CustomRetryError)
def custom_retry_error():
    raise Exception("Failed")

Q3: How to disable retries?

# Method 1: Use conditional decorators
import os
if os.getenv('ENABLE_RETRY', 'true').lower() == 'true':
    retry_decorator = retry(stop=stop_after_attempt(3))
else:
    retry_decorator = lambda func: func
@retry_decorator
def conditional_retry():
    pass

Q4: How to handle different exception types?

from tenacity import retry, retry_if_exception_type, stop_after_attempt
@retry(
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
    stop=stop_after_attempt(3))
def selective_retry():
    # Only specific exceptions will be retried
    pass

🧪 Best Practices

1. Retry Configuration Management

from tenacity import retry, stop_after_attempt, wait_exponential
from functools import wraps
def create_retry_config(max_attempts=3, base_wait=1, max_wait=10):
    """Create retry configuration"""
    return retry(
        stop=stop_after_attempt(max_attempts),
        wait=wait_exponential(multiplier=base_wait, min=base_wait, max=max_wait)
    )
# Use configuration
@create_retry_config(max_attempts=5, base_wait=2, max_wait=30)
def configured_retry():
    pass

2. Retry Monitoring

import logging
from tenacity import retry, stop_after_attempt, before_sleep_log
logger = logging.getLogger(__name__)
@retry(
    stop=stop_after_attempt(3),
    before_sleep=before_sleep_log(logger, logging.DEBUG))
def monitored_retry():
    logger.info("Executing retry operation")
    raise Exception("Needs retry")

3. Conditional Retry

from tenacity import retry, retry_if_result, stop_after_attempt
def should_retry(result):
    """Determine if a retry is needed"""
    return result is None or result.get('status') == 'pending'
@retry(
    retry=retry_if_result(should_retry),
    stop=stop_after_attempt(5))
def conditional_retry():
    # Decide whether to retry based on return value
    return {'status': 'pending'}

📚 References

Tenacity Official Documentation
GitHub Project Page
PyPI Page

📝 Summary

Tenacity provides a modern, elegant, powerful, and flexible way to control retries, making it an indispensable tool for building robust systems.

🌟 Core Advantages

Advantage	Description
Simple and Easy to Use	Decorator syntax, clear code
Powerful Functionality	Rich retry strategies and configuration options
Asynchronous Support	Full support for retrying asynchronous functions
Type Safety	Full support for type hints
Highly Customizable	Supports custom retry conditions and hook functions

Recommendation: The first choice for modern retry libraries, making error handling elegant and powerful!

🎯 Quick Start

# 1. Install
pip install tenacity
# 2. Basic usage
python -c "from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def demo():
    print('Trying...')
    raise Exception('Failed')
try:
    demo()
except Exception as e:
    print(f'Final failure: {e}')"

Python Daily Library: Tenacity – A Powerful Retry Library