-
Notifications
You must be signed in to change notification settings - Fork 1
/
retry.py
62 lines (48 loc) · 2 KB
/
retry.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
# Copyright 2024 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0
from restclients_core.exceptions import DataFailureException
import math
import time
def retry(ExceptionToCheck, tries=4, delay=3, backoff=2, status_codes=[],
logger=None):
"""
Decorator function for retrying the decorated function,
using an exponential or fixed backoff.
Original: https://wiki.python.org/moin/PythonDecoratorLibrary#Retry
ExceptionToCheck: the exception to check. Can be a tuple of
exceptions to check
tries: number of times to try (not retry) before giving up
delay: initial delay between tries in seconds
backoff: backoff multiplier
status_codes: list of http status codes to check for retrying, only applies
when ExceptionToCheck is a DataFailureException
logger: logging.Logger instance
"""
if backoff is None or backoff <= 0:
raise ValueError("backoff must be a number greater than 0")
tries = math.floor(tries)
if tries < 0:
raise ValueError("tries must be a number 0 or greater")
if delay is None or delay <= 0:
raise ValueError("delay must be a number greater than 0")
def deco_retry(f):
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay
while mtries > 1:
try:
return f(*args, **kwargs)
except ExceptionToCheck as err:
if (type(err) is DataFailureException and
len(status_codes) and
err.status not in status_codes):
raise
if logger:
logger.warning(
"{}: {}, Retrying in {} seconds.".format(
f.__name__, err, mdelay))
time.sleep(mdelay)
mtries -= 1
mdelay *= backoff
return f(*args, **kwargs)
return f_retry
return deco_retry