Skip to content

Instantly share code, notes, and snippets.

@rodelrod
Created September 24, 2015 13:44
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save rodelrod/ae6a1db487bfe089cc7b to your computer and use it in GitHub Desktop.
Save rodelrod/ae6a1db487bfe089cc7b to your computer and use it in GitHub Desktop.
Retry: Invokes repeatedly function calls that may raise an exception, until they don't
import time
def retry(f, tries, exc, pause=0, *args, **kwargs):
"""Repeats function n times until it succeeds or we give up.
Use case: `f` invokes an API that may fail or be unavailable. We assume
`f` raises an exception when something goes wrong. We can use this
function to invoke `f` multiple times until an exception is not raised.
Args:
f (callable): function to be called
tries (int): number of times the function is executed before we give up.
exc (Exception): exception class that is checked and results in a
retry if raised. If any other kind of exception is raised, it is
deliberately not caught because it's unexpected.
pause (int): seconds of pause between calls
*args, **kwargs: optional arguments passed on to f
Returns:
Return value of f if it succeeds in one of the tries.
Raises:
The passed exception exc if the calls to f fail every time with this
same exception. If the calls fail with an exception other than exc,
"""
last_try = tries - 1
for i in range(tries):
try:
result = f(*args, **kwargs)
return result
except exc:
if i == last_try:
# We give up
raise exc
else:
# Wait and then try again
time.sleep(pause)
continue
import unittest
from retry import retry
class BogusException(Exception):
pass
class Failer(object):
"""Simulates failing resource
Usage:
1. Instantiate Failer with the number of tries before the call succeeds:
>>> f = Failer(3)
2. Now, for the first n calls of f(), a BogusException is raised
3. On the (n+1)th call of f(), True is returned
"""
def __init__(self, n):
self.iterator = fail_n(n).next
def __call__(self):
if not self.iterator():
raise BogusException
else:
return True
def fail_n(n):
"""Generator yielding False n times and then True."""
for i in range(n):
yield False
while True:
yield True
class TestRetry(unittest.TestCase):
def test_retries_3_but_succeeds_too_late_on_5(self):
self.assertRaises(
BogusException,
retry, Failer(4), 3, BogusException,
)
def test_retries_4_and_succeeds_on_4(self):
self.assertEqual(
retry(Failer(3), 4, BogusException),
True
)
def test_retries_3_and_succeds_immediately(self):
self.assertEqual(
retry(Failer(1), 3, BogusException),
True
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment