debugpythonMinor
Python decorator for retrying w/exponential backoff
Viewed 0 times
retryingexponentialforpythonbackoffdecorator
Problem
This is my first decorator in Python! I found some of it on the internet but have tweaked it to our needs.
Here is the couple concerns of mine:
Here is the decorator.
```
import logging
import time
def retry_and_catch(exceptions, tries=5, logger=None, level=logging.ERROR, logger_attr=None, delay=0, backoff=0):
"""
Retries function up to amount of tries.
Backoff disabled by default.
:param exceptions: List of exceptions to catch
:param tries: Number of attempts before raising any exceptions
:param logger: Logger to print out to.
:param level: Log level.
:param logger_attr: Attribute on decorated class to get the logger ie self._logger you would give "_logger"
:param delay: initial delay seconds
:param backoff: backoff multiplier
"""
def deco_retry(f):
def f_retry(*args, **kwargs):
max_tries = tries
d = delay
exs = tuple(exceptions)
log = logger
while max_tries > 1:
try:
return f(*args, **kwargs)
except exs as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, max_tries)
# Get logger from cls instance of function
# Grabbing 'self'
instance = args[0]
if not log and logger_attr and hasattr(instance, logger_attr):
log = getattr(instance, logger_attr, None)
if log:
log.log(level, message)
else:
print(message)
# Sleep current delay
if d:
time.sleep(d)
# Increment delay
if backoff:
Here is the couple concerns of mine:
- Multiple python version compatibility
- Is grabbing the
selforarg[0]the best way to get the instance of the class?
- Any other general improvements!
Here is the decorator.
```
import logging
import time
def retry_and_catch(exceptions, tries=5, logger=None, level=logging.ERROR, logger_attr=None, delay=0, backoff=0):
"""
Retries function up to amount of tries.
Backoff disabled by default.
:param exceptions: List of exceptions to catch
:param tries: Number of attempts before raising any exceptions
:param logger: Logger to print out to.
:param level: Log level.
:param logger_attr: Attribute on decorated class to get the logger ie self._logger you would give "_logger"
:param delay: initial delay seconds
:param backoff: backoff multiplier
"""
def deco_retry(f):
def f_retry(*args, **kwargs):
max_tries = tries
d = delay
exs = tuple(exceptions)
log = logger
while max_tries > 1:
try:
return f(*args, **kwargs)
except exs as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, max_tries)
# Get logger from cls instance of function
# Grabbing 'self'
instance = args[0]
if not log and logger_attr and hasattr(instance, logger_attr):
log = getattr(instance, logger_attr, None)
if log:
log.log(level, message)
else:
print(message)
# Sleep current delay
if d:
time.sleep(d)
# Increment delay
if backoff:
Solution
First of all, when I start writing decorators of more than moderate complexity, and especially if they take parameters, I usually transition to writing them as classes - I find that easier to reason about and understand.
Second of all, in 100% of decorators I've ever written, I've wanted the decorator to look like the wrapped function. To do this, just use
Next, I think you have some weirdness in your api. I would expect
Next, you've limited yourself to some pretty static forms of backoff by making it a number. Instead, make it be a generator. Then you can do something like this
and then in your decorator, it looks like this
This lets you add much more complex backoff algorithms as needed.
Lastly, I think your comments don't really add much value to the code - reading the code is self-explanatory. I also don't think that promoting the values to local variables is worthwhile, but given that this is somewhat time sensitive then it might be. I removed that, but YMMV.
My end result looked something like this:
As an aside, I wrote a somewhat similar decorator and asked about it here Memoizing decorator that can retry, but the backoff idea is pretty cool and I might incorporate that into mine.
Second of all, in 100% of decorators I've ever written, I've wanted the decorator to look like the wrapped function. To do this, just use
functools.wraps().Next, I think you have some weirdness in your api. I would expect
logger to default to some default logger, and I would expect logger_attr to only be used to override whatever that logger is. I also don't like that you're explicitly calling print in the decorator - instead you should always use the logger, and if they want one that just calls print, or if that is the default, you should provide that. As an additional note, if logger_attr is a string then you can simplify the implementation a bit to look like this logger = getattr(instance, attr_name, logger)
logger.log(level, message)Next, you've limited yourself to some pretty static forms of backoff by making it a number. Instead, make it be a generator. Then you can do something like this
def doubling_backoff(start):
if start == 0:
start = 1
yield start
while True:
start *= 2
yield start
def no_backoff(start):
while True:
yield startand then in your decorator, it looks like this
backoff_gen = backoff(delay)
while max_tries > 1:
try:
return f(*args, **kwargs)
except exceptions as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, max_tries)
instance = args[0]
logger = getattr(args[0], attr_name, logger)
logger.log(level, message)
time.sleep(delay)
delay = next(backoff_gen)
max_tries -= 1This lets you add much more complex backoff algorithms as needed.
Lastly, I think your comments don't really add much value to the code - reading the code is self-explanatory. I also don't think that promoting the values to local variables is worthwhile, but given that this is somewhat time sensitive then it might be. I removed that, but YMMV.
My end result looked something like this:
import time
import logging
import functools
class DefaultLogger(object):
def log(self, level, message):
print(message)
def doubling_backoff(start):
if start == 0:
start = 1
yield start
while True:
start *= 2
yield start
def no_backoff(start):
while True:
yield start
class RetryAndCatch(object):
def __init__(exceptions_to_catch, num_tries=5, logger=DefaultLogger(), log_level=logging.ERROR, logger_attribute='', delay=0, backoff=no_backoff)
self.exceptions = exceptions_to_catch
self.max_tries = num_tries
self.tries = num_tries
self.logger = logger
self.level = log_level
self.attr_name = logger_attribute
self.delay = delay
self.backoff = backoff
def __call__(self, f):
@functools.wraps(f)
def retrier(*args, **kwargs):
backoff_gen = self.backoff(delay)
try:
while self.tries > 1:
try:
return f(*args, **kwargs)
except self.exceptions as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, self.tries)
instance = args[0]
self.logger = getattr(args[0], self.attr_name, self.logger)
self.logger.log(self.level, message)
time.sleep(self.delay)
self.delay = next(backoff_gen)
self.tries -= 1
return f(*args, **kwargs)
finally:
self.tries = self.max_tries
return retrierAs an aside, I wrote a somewhat similar decorator and asked about it here Memoizing decorator that can retry, but the backoff idea is pretty cool and I might incorporate that into mine.
Code Snippets
logger = getattr(instance, attr_name, logger)
logger.log(level, message)def doubling_backoff(start):
if start == 0:
start = 1
yield start
while True:
start *= 2
yield start
def no_backoff(start):
while True:
yield startbackoff_gen = backoff(delay)
while max_tries > 1:
try:
return f(*args, **kwargs)
except exceptions as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, max_tries)
instance = args[0]
logger = getattr(args[0], attr_name, logger)
logger.log(level, message)
time.sleep(delay)
delay = next(backoff_gen)
max_tries -= 1import time
import logging
import functools
class DefaultLogger(object):
def log(self, level, message):
print(message)
def doubling_backoff(start):
if start == 0:
start = 1
yield start
while True:
start *= 2
yield start
def no_backoff(start):
while True:
yield start
class RetryAndCatch(object):
def __init__(exceptions_to_catch, num_tries=5, logger=DefaultLogger(), log_level=logging.ERROR, logger_attribute='', delay=0, backoff=no_backoff)
self.exceptions = exceptions_to_catch
self.max_tries = num_tries
self.tries = num_tries
self.logger = logger
self.level = log_level
self.attr_name = logger_attribute
self.delay = delay
self.backoff = backoff
def __call__(self, f):
@functools.wraps(f)
def retrier(*args, **kwargs):
backoff_gen = self.backoff(delay)
try:
while self.tries > 1:
try:
return f(*args, **kwargs)
except self.exceptions as e:
message = "Exception {} caught, retrying {} more times.".format(e.message, self.tries)
instance = args[0]
self.logger = getattr(args[0], self.attr_name, self.logger)
self.logger.log(self.level, message)
time.sleep(self.delay)
self.delay = next(backoff_gen)
self.tries -= 1
return f(*args, **kwargs)
finally:
self.tries = self.max_tries
return retrierContext
StackExchange Code Review Q#133310, answer score: 8
Revisions (0)
No revisions yet.