Created
September 24, 2015 13:44
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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