Saltycrane logo

SaltyCrane Blog

Notes on Python, Django, and web development on Ubuntu Linux

    

Trying out a Retry decorator in Python

The Python wiki has a Retry decorator example which retries calling a failure-prone function using an exponential backoff algorithm. I modified it slightly to check for exceptions instead of a False return value to indicate failure. Each time the decorated function throws an exception, the decorator will wait a period of time and retry calling the function until the maximum number of tries is used up. If the decorated function fails on the last try, the exception will occur unhandled.

import time
from functools import wraps


def retry(ExceptionToCheck, tries=4, delay=3, backoff=2, logger=None):
    """Retry calling the decorated function using an exponential backoff.

    http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
    original from: http://wiki.python.org/moin/PythonDecoratorLibrary#Retry

    :param ExceptionToCheck: the exception to check. may be a tuple of
        exceptions to check
    :type ExceptionToCheck: Exception or tuple
    :param tries: number of times to try (not retry) before giving up
    :type tries: int
    :param delay: initial delay between retries in seconds
    :type delay: int
    :param backoff: backoff multiplier e.g. value of 2 will double the delay
        each retry
    :type backoff: int
    :param logger: logger to use. If None, print
    :type logger: logging.Logger instance
    """
    def deco_retry(f):

        @wraps(f)
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck, e:
                    msg = "%s, Retrying in %d seconds..." % (str(e), mdelay)
                    if logger:
                        logger.warning(msg)
                    else:
                        print msg
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            return f(*args, **kwargs)

        return f_retry  # true decorator

    return deco_retry

Try an "always fail" case

@retry(Exception, tries=4)
def test_fail(text):
    raise Exception("Fail")

test_fail("it works!")

Results:

Fail, Retrying in 3 seconds...
Fail, Retrying in 6 seconds...
Fail, Retrying in 12 seconds...
Traceback (most recent call last):
  File "retry_decorator.py", line 47, in 
    test_fail("it works!")
  File "retry_decorator.py", line 26, in f_retry
    f(*args, **kwargs)
  File "retry_decorator.py", line 33, in test_fail
    raise Exception("Fail")
Exception: Fail

Try a "success" case

@retry(Exception, tries=4)
def test_success(text):
    print "Success: ", text

test_success("it works!")

Results:

Success:  it works!

Try a "random fail" case

import random

@retry(Exception, tries=4)
def test_random(text):
    x = random.random()
    if x < 0.5:
        raise Exception("Fail")
    else:
        print "Success: ", text

test_random("it works!")

Results:

Fail, Retrying in 3 seconds...
Success:  it works!

Try handling multiple exceptions

Added 2010-04-27

import random

@retry((NameError, IOError), tries=20, delay=1, backoff=1)
def test_multiple_exceptions():
    x = random.random()
    if x < 0.40:
        raise NameError("NameError")
    elif x < 0.80:
        raise IOError("IOError")
    else:
        raise KeyError("KeyError")

test_multiple_exceptions()

Results:

IOError, Retrying in 1 seconds...
NameError, Retrying in 1 seconds...
IOError, Retrying in 1 seconds...
IOError, Retrying in 1 seconds...
NameError, Retrying in 1 seconds...
IOError, Retrying in 1 seconds...
NameError, Retrying in 1 seconds...
NameError, Retrying in 1 seconds...
NameError, Retrying in 1 seconds...
IOError, Retrying in 1 seconds...
Traceback (most recent call last):
  File "retry_decorator.py", line 61, in 
    test_multiple_exceptions("hello")
  File "retry_decorator.py", line 14, in f_retry
    f(*args, **kwargs)
  File "retry_decorator.py", line 56, in test_multiple_exceptions
    raise KeyError("KeyError")
KeyError: 'KeyError'

Unit tests

Added 2013-01-22. Note: Python 2.7 is required to run the tests.

import logging
import unittest

from decorators import retry


class RetryableError(Exception):
    pass


class AnotherRetryableError(Exception):
    pass


class UnexpectedError(Exception):
    pass


class RetryTestCase(unittest.TestCase):

    def test_no_retry_required(self):
        self.counter = 0

        @retry(RetryableError, tries=4, delay=0.1)
        def succeeds():
            self.counter += 1
            return 'success'

        r = succeeds()

        self.assertEqual(r, 'success')
        self.assertEqual(self.counter, 1)

    def test_retries_once(self):
        self.counter = 0

        @retry(RetryableError, tries=4, delay=0.1)
        def fails_once():
            self.counter += 1
            if self.counter < 2:
                raise RetryableError('failed')
            else:
                return 'success'

        r = fails_once()
        self.assertEqual(r, 'success')
        self.assertEqual(self.counter, 2)

    def test_limit_is_reached(self):
        self.counter = 0

        @retry(RetryableError, tries=4, delay=0.1)
        def always_fails():
            self.counter += 1
            raise RetryableError('failed')

        with self.assertRaises(RetryableError):
            always_fails()
        self.assertEqual(self.counter, 4)

    def test_multiple_exception_types(self):
        self.counter = 0

        @retry((RetryableError, AnotherRetryableError), tries=4, delay=0.1)
        def raise_multiple_exceptions():
            self.counter += 1
            if self.counter == 1:
                raise RetryableError('a retryable error')
            elif self.counter == 2:
                raise AnotherRetryableError('another retryable error')
            else:
                return 'success'

        r = raise_multiple_exceptions()
        self.assertEqual(r, 'success')
        self.assertEqual(self.counter, 3)

    def test_unexpected_exception_does_not_retry(self):

        @retry(RetryableError, tries=4, delay=0.1)
        def raise_unexpected_error():
            raise UnexpectedError('unexpected error')

        with self.assertRaises(UnexpectedError):
            raise_unexpected_error()

    def test_using_a_logger(self):
        self.counter = 0

        sh = logging.StreamHandler()
        logger = logging.getLogger(__name__)
        logger.addHandler(sh)

        @retry(RetryableError, tries=4, delay=0.1, logger=logger)
        def fails_once():
            self.counter += 1
            if self.counter < 2:
                raise RetryableError('failed')
            else:
                return 'success'

        fails_once()


if __name__ == '__main__':
    unittest.main()

Code / License

This code is also on github at: https://github.com/saltycrane/retry-decorator. It is BSD licensed.

6 Comments — feed icon Comments feed for this post


#1 Tiago Almeida commented on 2010-04-14:

This is great! Thanks for pointing this out. However i would modify it to return what the decorated function returns and throw (raise) the last exception given by the decorated function. Like this:

def retry(ExceptionToCheck, tries=4, delay=3, backoff=2):
    """Retry decorator
    original from http://wiki.python.org/moin/PythonDecoratorLibrary#Retry
    """
    def deco_retry(f):
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 0:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck, e:
                    print "%s, Retrying in %d seconds..." % (str(e), mdelay)
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
                    lastException = e
            raise lastException
        return f_retry # true decorator
    return deco_retry

#2 Eliot commented on 2010-04-14:

Tiago: Thanks for the comment! I tried out your modified decorator just now.

Raising the exception as you do is more elegant than my method, but I don't like the new behavior. Before raising the exception the last time, it will print "Retrying in xx seconds..." and then sleep for xx seconds (without actually retrying the function).

Making it return the value that the decorated function returns is a good idea. I updated my post above.


#3 Jaskirat commented on 2011-01-14:

In your code the lines after "return f(args, *kwargs)" will never be called. These lines seem to be unreachable code:

try_one_last_time = False
break

You might want to remove them


#4 ppa commented on 2012-08-21:
def retry(ExceptionToCheck, tries = 3, delay = 0.5, backoff = 1, logger = None):
    def deco_retry(f):
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck, e:
                    msg = "%s, Retrying in %d seconds..." % (str(e), mdelay)
                    if logger:
                        logger.warning(msg)
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff

            return f(*args, **kwargs)

        return f_retry  # true decorator
    return deco_retry

#5 Guillermo Siliceo commented on 2012-12-11:

Thanks for this! I added a little hook at the end to return a default and not an exception if all fails.

def retry(ExceptionToCheck, default=None, tries=4, delay=3, backoff=2, logger=None):
    """Retry calling the decorated function using an exponential backoff.

    http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
    original from: http://wiki.python.org/moin/PythonDecoratorLibrary#Retry

    :param ExceptionToCheck: the exception to check. may be a tuple of
        excpetions to check
    :type ExceptionToCheck: Exception or tuple
    :param tries: number of times to try (not retry) before giving up
    :type tries: int
    :param delay: initial delay between retries in seconds
    :type delay: int
    :param backoff: backoff multiplier e.g. value of 2 will double the delay
        each retry
    :type backoff: int
    :param logger: logger to use. If None, print
    :type logger: logging.Logger instance
    """
    def deco_retry(f):
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            try_one_last_time = True
            while mtries > 1:
                try:
                    print args,kwargs
                    return f(*args, **kwargs)
                    try_one_last_time = False
                    break
                except ExceptionToCheck, e:
                    msg = "%s, Retrying in %d seconds..." % (str(e), mdelay)
                    if logger:
                        logger.warning(msg)
                    else:
                        print msg
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            if try_one_last_time:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck, e:
                    return default
            return
        return f_retry  # true decorator
    return deco_retry

:)


#6 Eliot commented on 2013-01-22:

Jaskirat: I updated the code to remove the unreachable code. Thanks for catching that for me.

Post a comment

Required
Required, but not displayed
Optional

Format using Markdown. (No HTML.)
  • Code blocks: prefix each line by at least 4 spaces or 1 tab (and a blank line before and after)
  • Code span: surround with backticks
  • Blockquotes: prefix lines to be quoted with >
  • Links: <URL>
  • Links w/ description: [description](URL)
Created with Django | Hosted by Linode