Intermediate
You’ve built a Python script that calls a third-party API — weather data, a payment processor, a database over a flaky connection. Most of the time it works great. But occasionally the request times out, or the server returns a 503, or the network hiccups for half a second. Your script crashes. Your data pipeline fails at 3am. You get paged. Sound familiar?
The fix is retry logic — automatically re-attempting a failed operation before giving up. Python’s Tenacity library makes this trivially easy. It’s a standalone package (just pip install tenacity) that gives you a powerful @retry decorator and a rich set of stop conditions, wait strategies, and error callbacks — no need to write the same while True: try/except/sleep boilerplate ever again.
In this article we’ll cover how Tenacity works, how to configure it for real-world retry scenarios, and how to combine exponential backoff with jitter so your retries don’t make a thundering-herd problem worse. By the end, you’ll be able to wrap any flaky function with battle-tested retry logic in about five lines of code.
How To Use Python Tenacity: Quick Example
Here’s the simplest possible Tenacity usage — retry a function up to 3 times before giving up:
# quick_tenacity.py
import requests
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def fetch_status():
response = requests.get("https://httpbin.org/status/500")
response.raise_for_status() # raises HTTPError on 4xx/5xx
return response.status_code
try:
result = fetch_status()
print("Success:", result)
except Exception as e:
print("Failed after 3 attempts:", e)
Output:
Failed after 3 attempts: 500 Server Error: INTERNAL SERVER ERROR for url: https://httpbin.org/status/500
The @retry decorator intercepts the exception raised by raise_for_status(), waits 2 seconds, and tries again — up to 3 times total. When all attempts are exhausted, Tenacity re-raises the last exception. We’re using httpbin.org/status/500 here because it reliably returns a 500 error, making it a perfect test target without needing a broken server of your own.
That’s the core pattern. The rest of this article shows you how to customize the stop condition, wait strategy, and error handling for real production use.
What Is Tenacity and Why Use It?
Tenacity is a general-purpose retrying library for Python. It was forked from the older retrying package in 2016 and has been actively maintained since. Its key insight is that retry logic has several independent dimensions — when to stop, how long to wait between tries, which exceptions to retry on, and what to do when all tries are exhausted — and it lets you configure each dimension independently using composable building blocks.
Here’s how Tenacity compares to the most common alternatives:
| Approach | Pros | Cons |
|---|---|---|
Manual while loop | No dependencies | Repetitive, easy to get wrong, hard to test |
retrying (original) | Simple API | Unmaintained since 2016, no async support |
tenacity | Composable, async support, active maintenance | Small learning curve |
backoff | Minimal API | Fewer customization options than Tenacity |
Tenacity’s composable design means you can combine strategies freely — exponential backoff plus jitter plus a maximum delay cap, retrying only on specific exceptions, with a custom callback on each retry — all in one decorator call.
Controlling When to Stop Retrying
Tenacity gives you several composable stop conditions. You can combine them with the | operator to stop when either condition is met.
Stop After N Attempts
The most common stop condition: give up after a fixed number of tries. The count includes the first attempt — so stop_after_attempt(3) means 1 original call + 2 retries.
# stop_attempts.py
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(5))
def unstable_call():
raise ConnectionError("Server busy")
try:
unstable_call()
except Exception as e:
print(f"Gave up: {e}")
Output:
Gave up: Server busy
Stop After a Time Limit
Sometimes it’s more useful to cap the total elapsed time rather than the number of attempts. stop_after_delay(seconds) stops retrying once more than N seconds have elapsed since the first call.
# stop_delay.py
from tenacity import retry, stop_after_delay, wait_fixed
@retry(stop=stop_after_delay(10), wait=wait_fixed(3))
def slow_service():
raise TimeoutError("Still not ready")
try:
slow_service()
except Exception as e:
print(f"Timed out: {e}")
Output:
Timed out: Still not ready
This function retries every 3 seconds but will never run beyond 10 seconds total. The combination stop=stop_after_attempt(5) | stop_after_delay(30) stops whichever limit is hit first — a common pattern for production code.
Wait Strategies: Fixed, Exponential, and Jitter
How long you wait between retries matters as much as how many times you retry. Hammering a struggling server every millisecond makes things worse. Tenacity’s wait strategies let you control the delay precisely.
Fixed Wait
wait_fixed(seconds) waits the same amount of time between every attempt. Simple and predictable, good for testing or when the failure is likely transient and brief.
# wait_fixed_example.py
import time
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def ping():
print(f" Attempt at {time.strftime('%H:%M:%S')}")
raise IOError("No response")
try:
ping()
except Exception as e:
print(f"Failed: {e}")
Output:
Attempt at 09:00:00
Attempt at 09:00:01
Attempt at 09:00:02
Failed: No response
Exponential Backoff
Exponential backoff doubles the wait time after each failure. This is the right default for most API and network retries — it gives overloaded services breathing room rather than piling on more requests.
# wait_exponential_example.py
import time
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=1, max=30)
)
def call_api():
print(f" Attempt at {time.strftime('%H:%M:%S')}")
raise ConnectionError("Upstream unavailable")
try:
call_api()
except Exception as e:
print(f"Gave up: {e}")
Output:
Attempt at 09:00:00
Attempt at 09:00:01
Attempt at 09:00:03
Attempt at 09:00:07
Attempt at 09:00:15
Gave up: Upstream unavailable
The multiplier scales the base delay, min sets the floor (no delay shorter than 1s), and max caps the ceiling so delays don’t grow unboundedly on long retry sequences.
Adding Jitter
When many clients retry simultaneously after a service blip, exponential backoff alone creates “retry waves” — hundreds of requests hitting the server at the same moment. Adding random jitter spreads retries across time, preventing this thundering-herd effect. Use wait_random_exponential to combine both in one step:
# wait_jitter_example.py
from tenacity import retry, stop_after_attempt, wait_random_exponential
@retry(
stop=stop_after_attempt(6),
wait=wait_random_exponential(multiplier=1, max=60)
)
def distributed_fetch():
raise IOError("Service unavailable")
# Each client will retry on a slightly different schedule
The wait time is drawn from a random distribution bounded by the exponential cap, so two clients starting at the same time will take slightly different retry paths. AWS, Google Cloud, and Stripe all recommend this pattern in their API documentation.
Retrying on Specific Exceptions
By default Tenacity retries on any exception. In production you usually want to retry on transient errors (network timeouts, 503s) but not on permanent ones (401 Unauthorized, 404 Not Found, invalid input). Use retry_if_exception_type to be precise:
# retry_specific.py
import requests
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type
)
class TransientError(Exception):
pass
class PermanentError(Exception):
pass
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(TransientError)
)
def smart_fetch(url):
response = requests.get(url)
if response.status_code == 503:
raise TransientError("Service temporarily unavailable")
if response.status_code == 401:
raise PermanentError("Authentication failed -- check your API key")
response.raise_for_status()
return response.json()
try:
data = smart_fetch("https://httpbin.org/status/503")
except TransientError as e:
print(f"Gave up after retries: {e}")
except PermanentError as e:
print(f"Not retrying -- permanent error: {e}")
Output:
Gave up after retries: Service temporarily unavailable
You can also use retry_if_result to retry based on the return value rather than an exception — useful when a bad response returns 200 OK but with an error payload.
Before/After Callbacks and Logging
Tenacity supports before, after, and before_sleep callbacks for observability. The most useful is before_sleep, which fires before each wait period and lets you log retry attempts without cluttering the main function:
# retry_logging.py
import logging
from tenacity import (
retry, stop_after_attempt, wait_exponential,
before_sleep_log
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=1, min=2, max=20),
before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_with_logging():
raise ConnectionError("Upstream timeout")
try:
fetch_with_logging()
except ConnectionError as e:
logger.error("All retries exhausted: %s", e)
Output:
WARNING:__main__:Retrying fetch_with_logging in 2.0 seconds as it raised ConnectionError: Upstream timeout.
WARNING:__main__:Retrying fetch_with_logging in 4.0 seconds as it raised ConnectionError: Upstream timeout.
WARNING:__main__:Retrying fetch_with_logging in 8.0 seconds as it raised ConnectionError: Upstream timeout.
ERROR:__main__:All retries exhausted: Upstream timeout
The built-in before_sleep_log and after_log helpers format retry messages with the attempt number and wait time. You can also pass any callable to before_sleep that accepts a RetryCallState object for custom logging or metrics emission.
Async Support
Tenacity works with async def functions out of the box — the same decorator handles both sync and async functions. No separate import or special configuration needed:
# async_tenacity.py
import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
async def async_fetch(url: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status()
return response.json()
async def main():
try:
data = await async_fetch("https://jsonplaceholder.typicode.com/todos/1")
print("Title:", data["title"])
except Exception as e:
print("Failed:", e)
asyncio.run(main())
Output:
Title: delectus aut autem
We use httpx here because the standard requests library is synchronous and doesn’t play nicely with asyncio. If your project already uses aiohttp, Tenacity works the same way — just decorate the async def and it handles the rest.
Real-Life Example: Resilient API Client
Here’s a practical class that wraps an HTTP API with Tenacity retry logic, structured-data parsing, and proper error handling for both transient and permanent failures:
# resilient_client.py
import requests
import logging
from tenacity import (
retry, stop_after_attempt, wait_random_exponential,
retry_if_exception_type, before_sleep_log
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TransientAPIError(Exception):
"""Raised for 5xx responses and network errors -- safe to retry."""
pass
class PermanentAPIError(Exception):
"""Raised for 4xx responses -- do not retry."""
pass
class ResilientAPIClient:
BASE_URL = "https://jsonplaceholder.typicode.com"
@retry(
stop=stop_after_attempt(5),
wait=wait_random_exponential(multiplier=1, max=30),
retry=retry_if_exception_type(TransientAPIError),
before_sleep=before_sleep_log(logger, logging.WARNING)
)
def _get(self, path: str) -> dict:
try:
response = requests.get(self.BASE_URL + path, timeout=10)
except requests.exceptions.Timeout:
raise TransientAPIError("Request timed out")
except requests.exceptions.ConnectionError as e:
raise TransientAPIError(f"Connection failed: {e}")
if 500 <= response.status_code < 600:
raise TransientAPIError(f"Server error: {response.status_code}")
if 400 <= response.status_code < 500:
raise PermanentAPIError(f"Client error: {response.status_code}")
return response.json()
def get_user(self, user_id: int) -> dict:
return self._get(f"/users/{user_id}")
def get_posts_for_user(self, user_id: int) -> list:
return self._get(f"/posts?userId={user_id}")
if __name__ == "__main__":
client = ResilientAPIClient()
user = client.get_user(1)
print(f"User: {user['name']} ({user['email']})")
posts = client.get_posts_for_user(1)
print(f"Posts: {len(posts)} found")
for post in posts[:2]:
print(f" - {post['title'][:50]}")
Output:
User: Leanne Graham (Sincere@april.biz)
Posts: 10 found
- sunt aut facere repellat provident occaecati excepturi
- qui est esse
This pattern separates retry logic from business logic cleanly. The _get method knows only about HTTP mechanics and Tenacity. The get_user and get_posts_for_user methods know only about the API structure. To extend this, add methods for POST/PUT/DELETE operations (though be careful — only idempotent operations should be retried without caution).
Frequently Asked Questions
How do I install Tenacity?
Run pip install tenacity. No other dependencies are required. Tenacity supports Python 3.8+ and is actively maintained. To confirm it’s installed correctly, run python -c "import tenacity; print(tenacity.__version__)".
Does Tenacity re-raise the original exception?
Yes. When all attempts are exhausted, Tenacity raises a tenacity.RetryError by default, with the original exception stored in RetryError.__cause__. If you want the raw original exception raised instead, add reraise=True to the @retry decorator. Most production code uses reraise=True so the caller sees the actual failure, not a wrapper.
Is it safe to retry any function?
Only if the function is idempotent — meaning calling it twice has the same effect as calling it once. GET requests, read operations, and pure functions are safe. POST requests that create resources, payment operations, or any operation with side effects should NOT be retried blindly. For those, use Tenacity only on the specific exception types that confirm the operation did not go through (like a TimeoutError before the request completed).
Can I combine stop conditions?
Yes — use the | operator to stop when either condition is met: stop=stop_after_attempt(5) | stop_after_delay(60). This stops after 5 attempts OR after 60 seconds, whichever comes first. The & operator requires both conditions to be true simultaneously, which is less common but valid.
Does Tenacity work with async functions?
Yes, transparently. Apply @retry to an async def function and Tenacity detects the coroutine and awaits it correctly. The same wait strategies, stop conditions, and callbacks work for both sync and async code. There’s no separate import or class needed — the same from tenacity import retry handles both.
How do I test code that uses Tenacity in unit tests?
The cleanest approach is to patch the decorated function’s .retry attribute or mock the underlying function so it fails N times then succeeds. Alternatively, pass a custom stop=stop_after_attempt(1) in your test environment via the retry_with method: my_function.retry.stop = stop_after_attempt(1). This overrides the production retry count so tests run fast without sleeping.
Conclusion
Python Tenacity turns retry logic from a chore into a one-liner. We’ve covered the core decorator API, all the major stop conditions (stop_after_attempt, stop_after_delay), wait strategies (fixed, exponential, jitter), exception filtering with retry_if_exception_type, and built-in logging with before_sleep_log. The async support means you can use the same patterns in FastAPI, aiohttp, or any other async Python codebase.
The real-life example shows the most important architectural decision: keep your retry configuration on a single internal method and let higher-level methods stay clean. This makes it easy to tune retry parameters (attempt count, max delay) in one place as you learn how your upstream services actually behave.
For deeper customization — custom retry predicates, statistics callbacks, or Tenacity’s Retrying context-manager API — see the official Tenacity documentation.