Skip to content

Instantly share code, notes, and snippets.

@ar2pi
Last active March 20, 2023 15:38
Show Gist options
  • Save ar2pi/b0e0353f39405cdac09c6a21f1b9d72b to your computer and use it in GitHub Desktop.
Save ar2pi/b0e0353f39405cdac09c6a21f1b9d72b to your computer and use it in GitHub Desktop.
Retry with exponential backoff
#!/usr/bin/env python3
from contextlib import contextmanager
import functools
import math
import signal
import time
class TimeoutException(Exception):
pass
@contextmanager
def timeout(seconds):
def signal_handler(signum, frame):
raise TimeoutException(
f"Timed out after {seconds} seconds")
signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
def retry(wait_exponential_multiplier=1, wait_exponential_max=60, stop_max_delay=240, stop_max_attempt_number=math.inf):
"""
Retry with exponential backoff maxed at `wait_exponential_max`
until either `stop_max_delay` or `stop_max_attempt_number` is reached
Usage:
@retry()
some_unreliable_func():
pass
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_error = None
time_start = time.time()
time_since_start = 0
retry_count = 0
try:
with timeout(stop_max_delay):
while retry_count < stop_max_attempt_number:
try:
result = func(*args, **kwargs)
except Exception as e:
last_error = e
retry_count += 1
delay = min(wait_exponential_max,
wait_exponential_multiplier * 2**(retry_count - 1))
print(
f"Error: {e}. Retrying in {delay} seconds...")
time.sleep(delay)
else:
return result
except TimeoutException:
pass
time_since_start = time.time() - time_start
raise Exception(
f"Failed after {round(time_since_start, 2)}s, {retry_count} retries. Last error: {last_error}")
return wrapper
return decorator
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment