Source code for aspectlib.contrib

import time
from logging import getLogger

from aspectlib import Aspect

logger = getLogger(__name__)


[docs]def retry(func=None, retries=5, backoff=None, exceptions=(IOError, OSError, EOFError), cleanup=None, sleep=time.sleep): """ Decorator that retries the call ``retries`` times if ``func`` raises ``exceptions``. Can use a ``backoff`` function to sleep till next retry. Example:: >>> should_fail = lambda foo=[1,2,3]: foo and foo.pop() >>> @retry ... def flaky_func(): ... if should_fail(): ... raise OSError('Tough luck!') ... print("Success!") ... >>> flaky_func() Success! If it reaches the retry limit:: >>> @retry ... def bad_func(): ... raise OSError('Tough luck!') ... >>> bad_func() Traceback (most recent call last): ... OSError: Tough luck! """ @Aspect(bind=True) def retry_aspect(cutpoint, *args, **kwargs): for count in range(retries + 1): try: if count and cleanup: cleanup(*args, **kwargs) yield break except exceptions as exc: if count == retries: raise if not backoff: timeout = 0 elif isinstance(backoff, (int, float)): timeout = backoff else: timeout = backoff(count) logger.exception("%s(%s, %s) raised exception %s. %s retries left. Sleeping %s secs.", cutpoint.__name__, args, kwargs, exc, retries - count, timeout) sleep(timeout) return retry_aspect if func is None else retry_aspect(func)
[docs]def exponential_backoff(count): """ Wait 2**N seconds. """ return 2 ** count
retry.exponential_backoff = exponential_backoff
[docs]def straight_backoff(count): """ Wait 1, 2, 5 seconds. All retries after the 3rd retry will wait 5*N-5 seconds. """ return (1, 2, 5)[count] if count < 3 else 5 * count - 5
retry.straight_backoff = straight_backoff
[docs]def flat_backoff(count): """ Wait 1, 2, 5, 10, 15, 30 and 60 seconds. All retries after the 5th retry will wait 60 seconds. """ return (1, 2, 5, 10, 15, 30, 60)[count if count < 6 else -1]
retry.flat_backoff = flat_backoff