動機
很多時候，我們都喜歡為代碼加入retry功能。比如oauth驗證，有時候網絡不太靈，我們希望多試幾次。
這些retry應用的場景看起來不同，其實又很類似。都是判斷代碼是否正常運行，如果不是則重新開始。
那么，有沒有一種通用的辦法來實現呢？
簡介
Tenacity1是一個通用的retry庫，簡化為任何任務加入重試的功能。
它還包含如下特性:
- 通用的裝飾器API
- 可以設定重試停止的條件（比如設定嘗試次數）
- 可以設定重試間的等待時間（比如在嘗試之間使用冪數級增長的wait等待）
- 自定義在哪些Exception進行重試
- 自定義在哪些返回值的情況進行重試
- 協程的重試
用法
基本用法
from tenacity import *
# 基礎的用法，會一直重試下去，直到函數沒有拋出異常，正常返回值
@retry
def never_give_up_never_surrender():
print("一直重試，忽略exceptions，重試間沒有等待時間")
raise Exception
何時停止
讓我們加入停止的條件.
例如，在達到嘗試次數后停下來:
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("嘗試7次后停下")
raise Exception
在10秒后，如果仍然沒有成功，則停下:
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("10秒后停止")
raise Exception
可以使用|操作符，來組合多種條件:
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("10秒后，或者嘗試5次后，停下來")
raise Exception
嘗試間的等待
很多事並不是越快越好。所以，讓我們在重試的嘗試之間加入一些間隔時間:
@retry(wait=wait_fixed(2))
def wait_2_s():
print("每次重試間都有2秒間隔")
raise Exception
間隔可以是隨機的:
@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("重試間隔1-2秒")
raise Exception
還可以加入指數曲線形式的間隔:
@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1():
print("開始的時候等待 2^x * 1 秒，最少等待4秒，最多10秒，之后都是等待10秒")
raise Exception
多核在競爭一個共享的資源，使用指數間隔可以將沖突最小化:
@retry(wait=wait_random_exponential(multiplier=1, max=60))
def wait_exponential_jitter():
print("隨機等待 2^x * 1 秒，最多60秒，之后都是等待60秒")
raise Exception
可以自定義每次等待時長:
@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
[wait_fixed(7) for i in range(2)] +
[wait_fixed(9)]))
def wait_fixed_chained():
print("前三次等待3秒，后兩次等待7秒，最后一次等待9秒")
raise Exception
何時retry
默認情況下，只有函數拋出異常時才會retry。
你可以設置在制定的異常才進行retry：
@retry(retry=retry_if_exception_type(IOError))
def might_io_error():
print("只有在IOError的時候進行retry，其它時候照常拋出錯誤")
raise Exception
可以在判斷返回值是否是需要的情況下進行retry:
def is_none_p(value):
return value is None
@retry(retry=retry_if_result(is_none_p))
def might_return_none():
print("因為返回值是None，所以這個函數會一直retry")
# 這樣寫也是可以的，不用修改原來的代碼
retry_version_func = retry(retry=retry_if_result(is_none_p))(might_return_none)
當然，這里也可以組合多個條件:
def is_none_p(value):
return value is None
@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
print("在拋出任何異常，或者返回值是None的情況下，進行retry")
其它
在函數體內，你可以手動拋出TryAgain錯誤，進行重試:
@retry
def do_something():
result = something_else()
if result == 23:
raise TryAgain
通過參數reraise=True，可以拋出函數最后一次拋出的異常。如果沒有設定，會拋出RetryError:
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Fail")
try:
raise_my_exception()
except MyException:
print('MyException會被拋出')
在重試的前后，記錄日志:
import logging
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 raise_my_exception():
raise MyException("Fail")
# 重試后記錄
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Fail")
你可以獲取retry的相關統計數據:
@retry(stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Fail")
try:
raise_my_exception()
except Exception:
pass
print(raise_my_exception.retry.statistics)
熱度分析
這個庫已經6歲了，截止2019.5.4日已累計獲取1478star, 75fork.
源碼分析
這個庫在代碼和項目方面都是典范，同時API設計的也是相當漂亮。
這個庫對python裝飾器的用法已經爐火純青，基本所有的情景都有用到。有興趣的同學可以通過下面幾個點去看：
retry裝飾器為什么可以無參數版本/有參數版本混合使用retry裝飾器為什么可以作用函數和方法retry裝飾器為什么可以作用於asyncio協程，tornado協程，普通函數
個人評分
| 類型 | 評分 |
|---|---|
| 實用性 | ⭐️⭐️⭐️⭐️⭐️ |
| 易用性 | ⭐️⭐️⭐️⭐️⭐️ |
| 有趣性 | ⭐️⭐️⭐️ |
| 代碼質量 | ⭐️⭐️⭐️⭐️⭐️ |