-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathex_tenacity.py
More file actions
60 lines (48 loc) · 1.3 KB
/
ex_tenacity.py
File metadata and controls
60 lines (48 loc) · 1.3 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
# github : https://github.com/jd/tenacity
from tenacity import (
after_log,
before_log,
before_sleep_log,
retry_if_exception_type,
retry_if_not_exception_type,
retry_if_result,
retry,
stop_after_attempt,
stop_after_delay,
wait_exponential,
wait_fixed,
wait_random,
Retrying,
RetryError,
)
import logging
import sys
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
class MyError(Exception):
pass
@retry(
stop=(stop_after_attempt(max_attempt_number=5) | stop_after_delay(20)),
wait=wait_fixed(wait=5) + wait_random(min=0, max=5),
reraise=True,
before=before_log(logger, logging.DEBUG),
)
def func():
print("func executed")
raise MyError("my error in func ")
def func2(arg):
print(f"func2 executed, arg : {arg}")
raise MyError("my error in func2")
if __name__ == "__main__":
# case 1
func.retry_with(stop=stop_after_attempt(4))() # 런타임에서 옵션 변경
# case 2
retryer = Retrying(stop=stop_after_attempt(3), reraise=True)
retryer(func2, "test arg")
# case 3
try:
for attemp in Retrying(stop=stop_after_attempt(3)):
with attemp:
raise MyError("my error in code block")
except RetryError:
pass