-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdecorators.py
More file actions
71 lines (61 loc) · 1.98 KB
/
decorators.py
File metadata and controls
71 lines (61 loc) · 1.98 KB
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
63
64
65
66
67
68
69
70
71
import logging
import time
from functools import partial, wraps
def retry(func=None, exception=Exception, n_tries=5, delay=5, backoff=1, logger=False):
"""Retry decorator with exponential backoff.
Parameters
----------
func : typing.Callable, optional
Callable on which the decorator is applied, by default None
exception : Exception or tuple of Exceptions, optional
Exception(s) that invoke retry, by default Exception
n_tries : int, optional
Number of tries before giving up, by default 5
delay : int, optional
Initial delay between retries in seconds, by default 5
backoff : int, optional
Backoff multiplier e.g. value of 2 will double the delay, by default 1
logger : bool, optional
Option to log or print, by default False
Returns
-------
typing.Callable
Decorated callable that calls itself when exception(s) occur.
Examples
--------
>>> import random
>>> @retry(exception=Exception, n_tries=4)
... def test_random(text):
... x = random.random()
... if x < 0.5:
... raise Exception("Fail")
... else:
... print("Success: ", text)
>>> test_random("It works!")
"""
if func is None:
return partial(
retry,
exception=exception,
n_tries=n_tries,
delay=delay,
backoff=backoff,
logger=logger,
)
@wraps(func)
def wrapper(*args, **kwargs):
ntries, ndelay = n_tries, delay
while ntries > 1:
try:
return func(*args, **kwargs)
except exception as e:
msg = f"{str(e)}, Retrying in {ndelay} seconds..."
if logger:
logging.warning(msg)
else:
print(msg)
time.sleep(ndelay)
ntries -= 1
ndelay *= backoff
return func(*args, **kwargs)
return wrapper