Skip to content

Instantly share code, notes, and snippets.

@FBosler
Last active March 19, 2022 23:50
Show Gist options
  • Save FBosler/be10229aba491a8c912e3a1543bbc74e to your computer and use it in GitHub Desktop.
Save FBosler/be10229aba491a8c912e3a1543bbc74e to your computer and use it in GitHub Desktop.
retry.py
#Copyright 2021 Fabian Bosler
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy,
# modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom
# the Software is furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
# Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
# THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS
# OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from functools import wraps
import time
import logging
import random
logger = logging.getLogger(__name__)
def retry(exceptions, total_tries=4, initial_wait=0.5, backoff_factor=2, logger=None):
"""
calling the decorated function applying an exponential backoff.
Args:
exceptions: Exception(s) that trigger a retry, can be a tuple
total_tries: Total tries
initial_wait: Time to first retry
backoff_factor: Backoff multiplier (e.g. value of 2 will double the delay each retry).
logger: logger to be used, if none specified print
"""
def retry_decorator(f):
@wraps(f)
def func_with_retries(*args, **kwargs):
_tries, _delay = total_tries + 1, initial_wait
while _tries > 1:
try:
log(f'{total_tries + 2 - _tries}. try:', logger)
return f(*args, **kwargs)
except exceptions as e:
_tries -= 1
print_args = args if args else 'no args'
if _tries == 1:
msg = str(f'Function: {f.__name__}\n'
f'Failed despite best efforts after {total_tries} tries.\n'
f'args: {print_args}, kwargs: {kwargs}')
log(msg, logger)
raise
msg = str(f'Function: {f.__name__}\n'
f'Exception: {e}\n'
f'Retrying in {_delay} seconds!, args: {print_args}, kwargs: {kwargs}\n')
log(msg, logger)
time.sleep(_delay)
_delay *= backoff_factor
return func_with_retries
return retry_decorator
def log(msg, logger=None):
if logger:
logger.warning(msg)
else:
print(msg)
def test_func(*args, **kwargs):
rnd = random.random()
if rnd < .2:
raise ConnectionAbortedError('Connection was aborted :(')
elif rnd < .4:
raise ConnectionRefusedError('Connection was refused :/')
elif rnd < .8:
raise ConnectionResetError('Guess the connection was reset')
else:
return 'Yay!!'
if __name__ == '__main__':
# wrapper = retry((ConnectionAbortedError), tries=3, delay=.2, backoff=1, logger=logger)
# wrapped_test_func = wrapper(test_func)
# print(wrapped_test_func('hi', 'bye', hi='ciao'))
wrapper_all_exceptions = retry(Exception, total_tries=2, logger=logger)
wrapped_test_func = wrapper_all_exceptions(test_func)
print(wrapped_test_func('hi', 'bye', hi='ciao'))
@lbunzl-r7
Copy link

First - AWESOME! I sure gonna use this one.

In the docstring: "exceptions: Exeption(s) that trigger a retry, can be a tuble"
I guess you meant "tuple", right?

@FBosler
Copy link
Author

FBosler commented Feb 25, 2021

First - AWESOME! I sure gonna use this one.

In the docstring: "exceptions: Exeption(s) that trigger a retry, can be a tuble"
I guess you meant "tuple", right?

@liorbunzl Thanks so much! Appreciate it :)

Your comment made me laugh :) Had to think about bubbles. But jupp, ofc you are right. Thx for pointing it out.

@MathewDominic
Copy link

Thanks for this code. Really helpful :)

One doubt, in line no 22, Is there any reason why you've set _tries = total_tries + 1. Can't it just be _tries= total_tries? And check while _tries > 0

@arthur-c
Copy link

Hi @FBosler,

I would like to reuse your code, do you have a open source license attached to it?

@FBosler
Copy link
Author

FBosler commented Mar 25, 2021

Hi @FBosler,

I would like to reuse your code, do you have a open source license attached to it?

@arthur-c Sure I added it :)

@arthur-c
Copy link

Thank you!

@dakindre
Copy link

nice when you randomly stumble onto your ex-managers code

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment