Python重試庫Retrying和Tenacity的實現(xiàn)
簡介
Retrying 是一個通用重試庫,用于簡化任何需要重試的任務,已不再維護,功能:
- 通用裝飾器
- 指定停止條件,如重試次數(shù)
- 指定等待條件,如重試間的指數(shù)回退休眠
- 自定義異常重試
- 自定義重試預期返回的結果
Tenacity 是上述庫的分支,修復了一些BUG,增加了新功能:
- 異步協(xié)程重試
- 上下文管理器重試
- 組合停止條件
推薦使用 Tenacity
安裝
pip install retrying pip install tenacity
Retrying
初試
重試無數(shù)次直至成功
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise Exception
else:
print(f'tried {n} times')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
# tried 1 times
# tried 2 times
# tried 4 times
# tried 5 times
# tried 13 times
最大重試次數(shù)
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry(stop_max_attempt_number=3) # 最大重試次數(shù)
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise Exception
else:
print(f'tried {n} times and success.')
n = 0 # 重置重試次數(shù)
for i in range(5):
try:
do_something_unreliable()
except Exception as e:
print(f'tried {n} times but failed.')
n = 0 # 重置重試次數(shù)
# tried 3 times but failed.
# tried 1 times and success.
# tried 0 times and success.
# tried 2 times and success.
# tried 3 times but failed.
最大重試時間
單位為毫秒
import time
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry(stop_max_delay=3000) # 最大重試時間為3s
def do_something_unreliable():
global n
n += 1
raise Exception
start = time.time()
try:
do_something_unreliable()
except:
pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 796060 times, pass 2.999951124191284 seconds
重試間隔
import time
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait_fixed=500) # 重試間隔0.5秒
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 0 times, pass 0.0 seconds
# tried 3 times, pass 1.538625955581665 seconds
# tried 1 times, pass 0.5115864276885986 seconds
# tried 1 times, pass 0.5024125576019287 seconds
# tried 0 times, pass 0.0 seconds
隨機間隔
import time
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait_random_min=500, wait_random_max=1000) # 重試間隔0.5-1秒
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 1 times, pass 0.7865383625030518 seconds
# tried 1 times, pass 0.5917379856109619 seconds
# tried 6 times, pass 4.129276990890503 seconds
# tried 0 times, pass 0.0 seconds
# tried 3 times, pass 2.2903735637664795 seconds
指數(shù)級重試間隔
x為重試次數(shù),如第一次重試間隔2s,第二次4s,第三次8s,第四次16s(不設封頂?shù)脑挘?/p>
import time
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000) # 指數(shù)級重試間隔=2^x*1000ms,10s封頂
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 0 times, pass 0.0 seconds
# tried 1 times, pass 2.0003767013549805 seconds
# tried 0 times, pass 0.0 seconds
# tried 0 times, pass 0.0 seconds
# tried 4 times, pass 24.02167558670044 seconds
自定義異常重試
只重試 IOError
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry(retry_on_exception=lambda x: isinstance(x, IOError)) # 自定義異常重試
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise IOError
else:
print(f'tried {n} times')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
# tried 5 times
# tried 2 times
# tried 3 times
# tried 6 times
# tried 2 times
只重試 IOError,其余拋出
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry(retry_on_exception=lambda x: isinstance(x, IOError), wrap_exception=True) # 自定義異常結果
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise IOError
else:
print(f'tried {n} times')
n = 0 # 重置重試次數(shù)
if random.randint(0, 2) > 1:
raise IndexError
for i in range(5):
try:
do_something_unreliable()
except Exception as e:
print(e)
# tried 4 times
# tried 5 times
# tried 2 times
# tried 6 times
# tried 1 times
# RetryError[Attempts: 2, Error:
# File "C:\Users\Administrator\Envs\test\lib\site-packages\retrying.py", line 200, in call
# attempt = Attempt(fn(*args, **kwargs), attempt_number, False)
# File "D:/mycode/xxx.py", line 18, in do_something_unreliable
# raise IndexError
# ]
自定義返回重試
返回為 None 則重試
import random
from retrying import retry
n = 0 # 記錄重試次數(shù)
@retry(retry_on_result=lambda x: x is None) # 自定義異常重試
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
return None
else:
return n
for i in range(5):
print(f'tried {do_something_unreliable()} times')
n = 0 # 重置重試次數(shù)
# tried 10 times
# tried 8 times
# tried 0 times
# tried 10 times
# tried 3 times
參數(shù)
| 參數(shù) | 功能 |
|---|---|
| 無 | 重試直至成功 |
| stop_max_attempt_number | 最大重試次數(shù) |
| stop_max_delay | 最大重試時間(毫秒) |
| wait_fixed | 重試間隔(毫秒) |
| wait_random_min 和 wait_random_max | 隨機重試間隔(毫秒) |
| wait_exponential_multiplier 和 wait_exponential_max | 指數(shù)級重試間隔(毫秒) |
| retry_on_exception | 自定義異常重試 |
| wrap_exception | 是否拋出其余重試 |
| retry_on_result | 自定義異常結果 |
Tenacity
初試
import random
from tenacity import retry
n = 0 # 記錄重試次數(shù)
@retry
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise Exception
else:
print(f'tried {n} times')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
# tried 1 times
# tried 0 times
# tried 4 times
# tried 7 times
# tried 3 times
最大重試次數(shù)
import random
from tenacity import retry, stop_after_attempt
n = 0 # 記錄重試次數(shù)
@retry(stop=stop_after_attempt(3)) # 最大重試次數(shù)
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise Exception
else:
print(f'tried {n} times and success.')
n = 0 # 重置重試次數(shù)
for i in range(5):
try:
do_something_unreliable()
except Exception as e:
print(f'tried {n} times but failed.')
n = 0 # 重置重試次數(shù)
# tried 1 times and success.
# tried 3 times but failed.
# tried 3 times but failed.
# tried 2 times and success.
# tried 2 times and success.
最大重試時間
單位為秒
import time
from tenacity import retry, stop_after_delay
n = 0 # 記錄重試次數(shù)
@retry(stop=stop_after_delay(3)) # 最大重試時間為3s
def do_something_unreliable():
global n
n += 1
raise Exception
start = time.time()
try:
do_something_unreliable()
except:
pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds
組合停止條件
通過操作符 | 組合多個停止條件
import time
from tenacity import retry, stop_after_attempt, stop_after_delay
n = 0 # 記錄重試次數(shù)
@retry(stop=stop_after_attempt(300000) | stop_after_delay(3)) # 最大重試30w次 或 最長重試3秒
def do_something_unreliable():
global n
n += 1
raise Exception
start = time.time()
try:
do_something_unreliable()
except:
pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds
重試間隔
import time
import random
from tenacity import retry, wait_fixed
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait=wait_fixed(0.5)) # 重試間隔0.5秒
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 1 times, pass 0.5027601718902588 seconds
# tried 2 times, pass 1.0299296379089355 seconds
# tried 0 times, pass 0.0 seconds
# tried 3 times, pass 1.5316619873046875 seconds
# tried 4 times, pass 2.0474536418914795 seconds
隨機間隔
import time
import random
from tenacity import retry, wait_random
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait=wait_random(min=0.5, max=1)) # 重試間隔0.5-1秒
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 3 times, pass 2.287365674972534 seconds
# tried 0 times, pass 0.0 seconds
# tried 2 times, pass 1.4969894886016846 seconds
# tried 0 times, pass 0.0 seconds
# tried 6 times, pass 4.51520299911499 seconds
同樣可以組合
import time
import random
from tenacity import retry, wait_fixed, wait_random
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait=wait_fixed(0.5) + wait_random(min=0.5, max=1)) # 重試間隔1-1.5秒
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 2 times, pass 2.9294729232788086 seconds
# tried 0 times, pass 0.0 seconds
# tried 0 times, pass 0.0 seconds
# tried 3 times, pass 3.8608667850494385 seconds
# tried 1 times, pass 1.4092319011688232 seconds
指數(shù)級重試間隔
x為重試次數(shù),最小1s,最大10s
import time
import random
from tenacity import retry, wait_exponential
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait=wait_exponential(multiplier=1, min=1, max=10)) # 指數(shù)級重試間隔=2^x*1s,最小1s,最大10s
def do_something_unreliable():
global n, start
if random.randint(0, 3) > 1:
n += 1
raise Exception
else:
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
start = time.time() # 更新開始時間
# tried 0 times, pass 0.0 seconds
# tried 6 times, pass 35.06491994857788 seconds
# tried 2 times, pass 3.013124942779541 seconds
# tried 1 times, pass 1.010573387145996 seconds
# tried 0 times, pass 0.0 seconds
重試間隔鏈
import time
import random
from tenacity import retry, wait_fixed, wait_chain
n = 0 # 記錄重試次數(shù)
start = time.time() # 開始時間
@retry(wait=wait_chain(*[wait_fixed(0.5)] * 3 + [wait_fixed(1)] * 2 + [wait_fixed(2)])) # 重試間隔鏈,先3個0.5秒,再2個1秒,之后都是2秒
def do_something_unreliable():
global n
if n == 10:
return
n += 1
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
raise Exception
do_something_unreliable()
# tried 1 times, pass 0.0 seconds
# tried 2 times, pass 0.5056586265563965 seconds
# tried 3 times, pass 1.0193772315979004 seconds
# tried 4 times, pass 1.5333683490753174 seconds
# tried 5 times, pass 2.5386297702789307 seconds
# tried 6 times, pass 3.5489938259124756 seconds
# tried 7 times, pass 5.551833629608154 seconds
# tried 8 times, pass 7.559761047363281 seconds
# tried 9 times, pass 9.561469554901123 seconds
# tried 10 times, pass 11.570155143737793 seconds
自定義異常重試
只重試 IOError
import random
from tenacity import retry, retry_if_exception_type
n = 0 # 記錄重試次數(shù)
@retry(retry=retry_if_exception_type(IOError)) # 自定義異常重試
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
raise IOError
else:
print(f'tried {n} times')
n = 0 # 重置重試次數(shù)
for i in range(5):
do_something_unreliable()
# tried 5 times
# tried 2 times
# tried 3 times
# tried 6 times
# tried 2 times
自定義返回重試
返回為 None 則重試
import random
from tenacity import retry, retry_if_result
n = 0 # 記錄重試次數(shù)
@retry(retry=retry_if_result(lambda x: x is None)) # 自定義異常結果
def do_something_unreliable():
global n
if random.randint(0, 10) > 1:
n += 1
return None
else:
return n
for i in range(5):
print(f'tried {do_something_unreliable()} times')
n = 0 # 重置重試次數(shù)
# tried 10 times
# tried 8 times
# tried 0 times
# tried 10 times
# tried 3 times
顯式重試
from tenacity import retry, stop_after_delay, TryAgain
@retry(stop=stop_after_delay(3))
def do_something_unreliable(n):
"""顯式重試"""
n += 1
if n == 10:
raise TryAgain
return n
n = 0
while n <= 15:
try:
n = do_something_unreliable(n)
except:
n += 1
else:
print(n, end=' ')
# 1 2 3 4 5 6 7 8 9 11 12 13 14 15 16
重新捕獲異常
超過最大重試次數(shù)或時間后拋出異常
import time
from tenacity import retry, stop_after_attempt, stop_after_delay
n = 0
start = time.time()
@retry(reraise=True, stop=stop_after_attempt(3))
def do_something_unreliable():
"""超過最大重試次數(shù)后拋出異常"""
global n
n += 1
raise IOError(f'tried {n} times but failed.')
@retry(reraise=True, stop=stop_after_delay(3))
def do_something_unreliable1():
"""超過最大重試時間后拋出異常"""
end = time.time()
raise IOError(f'tried {end - start:.4f} seconds but failed.')
try:
do_something_unreliable()
except Exception as e:
print(e)
try:
do_something_unreliable1()
except Exception as e:
print(e)
# tried 3 times but failed.
# tried 2.9864 seconds but failed.
重試前記日志
import sys
import logging
from tenacity import retry, stop_after_attempt, before_log
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def do_something_unreliable():
raise IOError('Fail')
try:
do_something_unreliable()
except Exception as e:
print(e)
# RetryError[<Future at 0x1b0342d7a58 state=finished raised OSError>]
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 1st time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 2nd time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 3rd time calling it.
重試后記日志
import sys
import time
import logging
from tenacity import retry, stop_after_attempt, after_log
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def do_something_unreliable():
time.sleep(0.5)
raise IOError('Fail')
try:
do_something_unreliable()
except Exception as e:
print(e)
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 0.500(s), this was the 1st time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.000(s), this was the 2nd time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.500(s), this was the 3rd time calling it.
# RetryError[<Future at 0x22b45e07a58 state=finished raised OSError>]
重試失敗后記日志
import sys
import time
import logging
from tenacity import retry, stop_after_attempt, before_sleep_log
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before_sleep=before_sleep_log(logger, logging.DEBUG))
def do_something_unreliable():
time.sleep(0.5)
raise IOError('Fail')
try:
do_something_unreliable()
except Exception as e:
print(e)
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# RetryError[<Future at 0x1c840b96ac8 state=finished raised OSError>]
重試統(tǒng)計
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def do_something_unreliable():
raise IOError('Fail')
try:
do_something_unreliable()
except:
pass
print(do_something_unreliable.retry.statistics)
# {'start_time': 756545.281, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0}
do_something_unreliable.retry.statistics["attempt_number"] 可直接獲取重試次數(shù),不用手動計算
自定義回調(diào)函數(shù)
retry_state 是 RetryCallState 的實例
from tenacity import retry, stop_after_attempt, retry_if_result
n = 0
def return_value(retry_state):
"""自定義回調(diào)函數(shù)"""
return retry_state.outcome.result() # 返回函數(shù)產(chǎn)生的最后結果或異常
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_value,
retry=retry_if_result(lambda x: isinstance(x, int)))
def do_something_unreliable():
global n
n += 1
return n
print(do_something_unreliable())
# 3
上下文管理器
結合 for 循環(huán)和上下文管理器
from tenacity import Retrying, RetryError, stop_after_attempt
try:
for attempt in Retrying(stop=stop_after_attempt(3)):
with attempt:
raise Exception('Fail')
except RetryError:
pass
協(xié)程
from tenacity import AsyncRetrying, RetryError, stop_after_attempt
async def function():
try:
async for attempt in AsyncRetrying(stop=stop_after_attempt(3)):
with attempt:
raise Exception('Fail')
except RetryError:
pass
支持異步和協(xié)程
import trio
import asks
import tornado
from tenacity import retry
@retry
async def my_async_function(loop):
await loop.getaddrinfo('8.8.8.8', 53)
@retry
@tornado.gen.coroutine
def my_async_function(http_client, url):
yield http_client.fetch(url)
@retry(sleep=trio.sleep)
async def my_async_function(loop):
await asks.get('https://example.org')
參數(shù)
| 參數(shù) | 功能 | 取值 |
|---|---|---|
| 無 | 重試直至成功 | |
| stop | 停止條件 | 最大重試次數(shù) stop_after_attempt 最大重試時間 stop_after_delay |
| wait | 等待條件 | 重試間隔 wait_fixed 隨機重試間隔wait_random 指數(shù)級重試間隔 wait_exponential 重試間隔鏈 wait_chain |
| retry | 重試條件 | 自定義異常 retry_if_exception_type 自定義返回 retry_if_result |
| reraise | 是否拋出異常 | bool |
| before | 重試前動作 | 重試前記日志 before_log |
| after | 重試后動作 | 重試后記日志 after_log |
| before_sleep | 重試失敗后動作 | before_sleep_log 重試失敗后記日志 |
| retry_error_callback | 回調(diào)函數(shù) |
到此這篇關于Python重試庫Retrying和Tenacity的實現(xiàn)的文章就介紹到這了,更多相關Python重試庫Retrying和Tenacity內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
Python數(shù)據(jù)清洗工具之Numpy的基本操作
Numpy的操作對象是一個ndarray,所以在使用這個庫進行計算的時候需要將數(shù)據(jù)進行轉化,這篇文章主要介紹了Python數(shù)據(jù)清洗工具之Numpy的基本操作,需要的朋友可以參考下2021-04-04
Python連接MySQL數(shù)據(jù)庫連接池的操作詳解
連接池的優(yōu)點是可以在多個線程或進程之間共享,并且可以有效地管理連接數(shù),而無需手動打開和關閉連接,下面我們就來看看Python如何連接MySQL數(shù)據(jù)庫連接池吧2025-05-05

