-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathretrying.py
More file actions
66 lines (59 loc) · 1.55 KB
/
retrying.py
File metadata and controls
66 lines (59 loc) · 1.55 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
"""
A redesign of
http://code.activestate.com/recipes/578527-retry-loop/?in=lang-python
Its example is like example() except its version would start:
for retry in retryloop(3, timeout=2):
"""
def example():
for retry in retrying(timeout(2, range(3))):
try:
print 'something'
# time.sleep(1) # Uncomment to test timeout
assert False, "No go, Mo"
except AssertionError:
retry()
import time
def retrying(trials):
def retry():
retry.done = False
for trial in trials:
retry.done = True
yield retry
if retry.done:
return
raise RetryError
class RetryError(Exception): pass
def timeout(interval, trials):
deadline = time.time() + interval
for trial in trials:
yield trial
if deadline <= time.time(): break
## example()
#. something
#. something
#. something
#.
#. Traceback (most recent call last):
#. File "retrying.py", line 9, in example
#. for retry in retrying(timeout(2, range(3))):
#. File "retrying.py", line 27, in retrying
#. raise RetryError
#. RetryError
def fancy_retrying(trials):
def retry(exc=None):
retry.done = False
retry.exc = exc
retry.exc = None
trial = None
for trial in trials:
retry.done = True
yield retry
if retry.done:
return
# All the trials failed; complain.
info = "on trial {0}".format(trial)
if retry.exc:
retry.exc.args += (info,)
raise retry.exc
else:
raise RetryError(info)