tenacity
tenacity 9.0.0
請參閱堅韌文件以獲得更好的體驗。
Tenacity 是一個 Apache 2.0 許可的通用重試函式庫，用 Python 編寫，用於簡化向任何事物添加重試行為的任務。它源自於重試分叉，遺憾的是該分叉已不再維護。 Tenacity 與重試的 api 不相容，但添加了重要的新功能並修復了許多長期存在的錯誤。
最簡單的用例是每當發生異常時重試片狀函數，直到返回值。
..測試程式碼::
隨機導入
從堅韌導入重試
@重試
def do_something_unreliable():
如果 random.randint(0, 10) > 1:
raise IOError("醬汁壞了，一切都被沖掉了!!!111one")
別的：
返回“很棒的醬汁！”
列印（do_something_unreliable（））
..測試輸出:: ：隱藏： 很棒的醬汁！
.. 目錄樹::
：隱：
：最大深度：2
變更日誌
應用程式介面
要安裝tenacity ，只需：
$ pip install tenacity..測試設定::
匯入日誌記錄
#
# 注意以下導入僅用於演示方便。
# 生產代碼應始終明確匯入所需的名稱。
#
來自堅韌進口*
類別 MyException（異常）：
經過
正如您在上面看到的，預設行為是在引發異常時永遠重試而不等待。
..測試程式碼::
@重試
def never_gonna_give_you_up():
print("永遠重試，忽略異常，不要在重試之間等待")
引發例外
讓我們少一些執著，設定一些界限，例如放棄之前的嘗試次數。
..測試程式碼::
@重試（停止=嘗試後停止（7））
def stop_after_7_attempts():
print("嘗試7次後停止")
引發例外
我們沒有一整天的時間，所以讓我們為應該重試的時間設定一個界限。
..測試程式碼::
@重試（停止= stop_after_delay（10））
def stop_after_10_s():
print("10秒後停止")
引發例外
如果您的截止日期很緊，並且不能超過延遲時間，那麼您可以在超過延遲時間之前放棄重試一次。
..測試程式碼::
@重試（停止= stop_before_delay（10））
def stop_before_10_s():
print("10 秒前停止 1 次嘗試")
引發例外
您可以使用 | 組合多個停止條件操作員：
..測試程式碼::
@重試（停止=（stop_after_delay（10）| stop_after_attempt（5）））
def stop_after_10_s_or_5_retries():
print("10秒或重試5次後停止")
引發例外
大多數事情不希望盡快輪詢，因此我們在重試之間等待 2 秒。
..測試程式碼::
@重試（等待= wait_fixed（2））
def wait_2_s():
print("重試之間等待 2 秒")
引發例外
有些事情在註入一點隨機性時表現最好。
..測試程式碼::
@重試（等待= wait_random（最小值= 1，最大值= 2））
def wait_random_1_to_2_s():
print("重試之間隨機等待 1 到 2 秒")
引發例外
但話又說回來，在重試分散式服務和其他遠端端點時，很難擊敗指數退避。
..測試程式碼::
@重試（等待= wait_exponential（乘數= 1，最小值= 4，最大值= 10））
def wait_exponential_1():
print("每次重試之間等待 2^x * 1 秒，從 4 秒開始，然後最多 10 秒，然後再等待 10 秒")
引發例外
話又說回來，在重試分散式服務和其他遠端端點時，將固定等待和抖動（以幫助避免驚群）相結合也很難被擊敗。
..測試程式碼::
@retry(等待=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("等待至少3秒，並添加最多2秒的隨機延遲")
引發例外
當多個進程爭奪共享資源時，指數增長的抖動有助於最大限度地減少衝突。
..測試程式碼::
@重試（等待= wait_random_exponential（乘數= 1，最大值= 60））
def wait_exponential_jitter():
print("每次重試之間隨機等待最多 2^x * 1 秒，直到範圍達到 60 秒，然後隨機等待 60 秒")
引發例外
有時有必要建立一個退避鏈。
..測試程式碼::
@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
[wait_fixed(7) for i in range(2)] +
[等待固定(9)]))
def wait_fixed_chained():
print("3 次嘗試等待 3 秒，接下來的 2 次嘗試等待 7 秒，此後的所有嘗試等待 9 秒")
引發例外
我們有幾種選擇來處理引發特定或一般異常的重試，就像這裡的情況一樣。
..測試程式碼::
類別客戶端錯誤（異常）：
"""某種類型的客戶端錯誤。"""
@retry(重試=retry_if_exception_type(IOError))
def might_io_error():
print("如果發生 IOError，則永遠重試，無需等待，引發任何其他錯誤")
引發例外
@retry(重試=retry_if_not_exception_type(ClientError))
def might_client_error():
print("如果發生 ClientError 以外的任何錯誤，則永遠重試，無需等待。立即引發 ClientError。")
引發例外
我們也可以使用函數的結果來改變重試的行為。
..測試程式碼::
def is_none_p(值):
"""如果值為 None，則傳回 True"""
傳回值為無
@retry(重試=retry_if_result(is_none_p))
def might_return_none():
print("如果傳回值為 None，則不等待重試")
另請參閱這些方法：
..測試程式碼::
如果異常則重試
如果異常類型則重試
如果不異常則重試類型
重試除非異常類型
如果結果重試
如果沒有結果則重試
重試如果_異常_訊息
如果不異常則重試訊息
重試任意
全部重試
我們也可以組合幾個條件：
..測試程式碼::
def is_none_p(值):
"""如果值為 None，則傳回 True"""
傳回值為無
@retry(重試=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
print("如果傳回值為 None，永遠重試，忽略異常，無需等待")
也支援停止、等待等任意組合，讓您可以自由地混合和匹配。
還可以透過引發 TryAgain 異常來隨時明確重試：
..測試程式碼::
@重試
def do_something():
結果 = some_else()
如果結果 == 23：
提出再試一次
通常，當您的函數最後一次失敗（並且不會根據您的設定再次重試）時，會引發 RetryError。您的程式碼遇到的異常將顯示在堆疊追蹤中間的某個位置。
如果您希望在堆疊追蹤末端（最明顯的位置）看到程式碼遇到的異常，則可以設定 reraise=True。
..測試程式碼::
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
引發 MyException(“失敗”)
嘗試：
引發我的異常（）
除了我的異常：
# 重試超時
經過
透過使用 before 回呼函數，可以在嘗試呼叫函數之前執行一個操作：
..測試程式碼::
匯入日誌記錄
導入系統
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
記錄器=logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger,logging.DEBUG))
def raise_my_exception():
引發 MyException(“失敗”)
本著同樣的精神，可以在呼叫失敗後執行：
..測試程式碼::
匯入日誌記錄
導入系統
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
記錄器=logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger,logging.DEBUG))
def raise_my_exception():
引發 MyException(“失敗”)
也可以只記錄將要重試的失敗。通常重試發生在等待間隔之後，因此關鍵字參數稱為before_sleep ：
..測試程式碼::
匯入日誌記錄
導入系統
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
記錄器=logging.getLogger(__name__)
@重試（停止=嘗試後停止（3），
before_sleep = before_sleep_log（記錄器，logging.DEBUG））
def raise_my_exception():
引發 MyException(“失敗”)
您可以使用函數附加的統計資訊屬性來存取有關函數重試的統計資料：
..測試程式碼::
@重試（停止=嘗試後停止（3））
def raise_my_exception():
引發 MyException(“失敗”)
嘗試：
引發我的異常（）
除了例外：
經過
列印（raise_my_exception.statistics）
..測試輸出:: ：隱藏： ……
您也可以定義自己的回調。回呼應該接受一個名為retry_state的參數，其中包含有關當前重試呼叫的所有資訊。
例如，您可以在所有重試失敗後呼叫自訂回調函數，而不會引發異常（或您可以重新引發或執行任何操作）
..測試程式碼::
def return_last_value(重試狀態):
"""傳回最後一次呼叫嘗試的結果"""
返回 retry_state.outcome.result()
def is_false(值):
"""如果值為 False，則傳回 True"""
傳回值為False
# 嘗試3次後得到不同的結果會回傳False
@重試（停止=嘗試後停止（3），
retry_error_callback=return_last_value,
重試=retry_if_result(is_false))
def Final_return_false():
回傳錯誤
retry_state參數是:class:`~tenacity.RetryCallState`類別的物件。
也可以為其他關鍵字參數定義自訂回調。
.. 函數:: my_stop(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息 :return: 重試是否應該停止 :r 類型：布林值
.. 函數:: my_wait(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息 :return: 下次重試前等待的秒數 :r類型：浮點數
.. 函數:: my_retry(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息 :return: 是否繼續重試 :r 類型：布林值
.. 函數:: my_before(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息
.. 函數:: my_after(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息
.. 函數:: my_before_sleep(retry_state) :param RetryCallState retry_state: 有關當前重試呼叫的信息
這是一個自訂before_sleep函數的範例：
..測試程式碼::
匯入日誌記錄
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
記錄器=logging.getLogger(__name__)
def my_before_sleep(retry_state):
如果 retry_state.attempt_number < 1：
日誌等級 = 日誌記錄.INFO
別的：
日誌等級 = 日誌記錄.警告
記錄器.log(
loglevel, '重試 %s: 嘗試 %s 結束於: %s',
retry_state.fn、retry_state.attempt_number、retry_state.outcome）
@retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
def raise_my_exception():
引發 MyException(“失敗”)
嘗試：
引發我的異常（）
除了重試錯誤：
經過
您可以在呼叫重試裝飾器時根據需要更改其參數，方法是使用附加到包裝函數的 retry_with 函數：
..測試程式碼::
@重試（停止=嘗試後停止（3））
def raise_my_exception():
引發 MyException(“失敗”)
嘗試：
raise_my_exception.retry_with(stop=stop_after_attempt(4))()
除了例外：
經過
列印（raise_my_exception.statistics）
..測試輸出:: ：隱藏： ……
如果你想使用變數來設定重試參數，則不必使用重試裝飾器 - 你可以直接使用 Retrying：
..測試程式碼::
def never_good_enough(arg1):
引發異常（'無效參數：{}'.format（arg1））
def try_never_good_enough(max_attempts=3):
重試器 = 重試(stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, '我真的嘗試過')
您可能還想暫時更改修飾函數的行為，例如在測試中以避免不必要的等待時間。您可以修改/修補附加到函數的重試屬性。請記住，這是一個只寫屬性，統計資訊應該從函數統計屬性中讀取。
..測試程式碼::
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def raise_my_exception():
引發 MyException(“失敗”)
從單元測試導入模擬
與mock.patch.object（raise_my_exception.retry，“等待”，wait_fixed（0））：
嘗試：
引發我的異常（）
除了例外：
經過
列印（raise_my_exception.statistics）
..測試輸出:: ：隱藏： ……
Tenacity 可讓您重試程式碼區塊，而無需將其包裝在獨立的函數中。這使得在共享上下文時可以輕鬆隔離故障塊。訣竅是將 for 循環和上下文管理器結合。
..測試程式碼::
from tenacity import 重試、重試錯誤、stop_after_attempt
嘗試：
對於重試的嘗試（stop=stop_after_attempt(3)）：
嘗試：
引發異常（'我的程式碼失敗！'）
除了重試錯誤：
經過
您可以透過設定 Retrying 物件來設定重試策略的各個細節。
對於非同步程式碼，您可以使用 AsyncRetrying。
..測試程式碼::
from tenacity import AsyncRetrying、RetryError、stop_after_attempt
非同步定義函數()：
嘗試：
AsyncRetrying 中的嘗試非同步（stop=stop_after_attempt(3)）：
嘗試：
引發異常（'我的程式碼失敗！'）
除了重試錯誤：
經過
在這兩種情況下，您可能想要將結果設定為嘗試，以便在retry_if_result等重試策略中可用。這可以透過存取retry_state屬性來完成：
..測試程式碼::
從堅韌導入AsyncRetrying，retry_if_result
非同步定義函數()：
非同步嘗試 AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
嘗試：
result = 1 # 一些複雜的計算、函數呼叫等。
如果不是嘗試.retry_state.outcome.failed：
attempts.retry_state.set_result(結果)
回傳結果
最後， retry也適用於 asyncio、Trio 和 Tornado (>= 4.5) 協程。睡眠也是異步完成的。
@ retry
async def my_asyncio_function ( loop ):
await loop . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
async def my_async_trio_function ():
await trio . socket . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
@ tornado . gen . coroutine
def my_async_tornado_function ( http_client , url ):
yield http_client . fetch ( url )您甚至可以透過傳遞正確的睡眠函數來使用替代事件循環，例如 curio：
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno 用於管理變更日誌。看看他們的使用文件。
文件產生將自動編譯變更日誌。您只需添加它們即可。
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
