Skip to content

Instantly share code, notes, and snippets.

@edvardm
Created July 27, 2021 14:57
Show Gist options
  • Save edvardm/8b99494611bb6fb016608c15c4a1b731 to your computer and use it in GitHub Desktop.
Save edvardm/8b99494611bb6fb016608c15c4a1b731 to your computer and use it in GitHub Desktop.
simple retry decorator with exp backoff
import time
import sys
def exp_backoff_retry(
fun=None,
*,
retry_exceptions=(BaseException,),
max_retries=3,
exp_base=2.0,
max_seconds=300,
):
"""
Simple timeout decorator. Use eg. backoff for more proper one
Kwargs:
retry_exceptions: tuple of exceptions to catch and retry, by default any exception causes retry
except NameError, SyntaxError and ImportError
max_retries: maximum number of retries
exp_base: base for exponent in backoff delay. Defaults to 2
max_seconds: maximum total time for retries in seconds. Defaults to 300
"""
def decor(orig_fun):
def log(msg):
sys.stderr.write(f"{msg}\n")
def inner(*args, **kwargs):
start_t = time.time()
retries = 0
while True:
try:
return orig_fun(*args, **kwargs)
except (NameError, SyntaxError, ImportError):
raise
except retry_exceptions as exc:
if retries >= max_retries:
log(f"Max retries of {max_retries} reached, raising")
raise
delay = exp_base ** retries
# If we have taken only 2 seconds, next sleep is for 4 seconds but max_seconds is 5,
# we can already give up as 2+4 > 5
if time.time() - start_t + delay > max_seconds:
log(f"Max time of {max_seconds:.3f} reached, raising")
raise
retries += 1
log(
f"Caught exception {exc}, attempt {retries}/{max_retries} in {delay:.3f}s"
)
time.sleep(delay)
return inner
if fun:
return decor(fun)
return decor
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment