您當前的位置:首頁 > 詩詞

少有人知的 Python "重試機制":tenacity

作者:由 王炳明 發表于 詩詞時間:2020-11-09

為了避免由於一些網路或等其他不可控因素,而引起的功能性問題。比如在傳送請求時,會因為網路不穩定,往往會有請求超時的問題。

這種情況下,我們通常會在程式碼中加入重試的程式碼。重試的程式碼本身不難實現,但如何寫得優雅、易用,是我們要考慮的問題。

這裡要給大家介紹的是一個第三方庫 -

Tenacity

,它實現了幾乎我們可以使用到的所有重試場景,比如:

在什麼情況下才進行重試?

重試幾次呢?

重試多久後結束?

每次重試的間隔多長呢?

重試失敗後的回撥?

在使用它之前 ,先要安裝它

$ pip install tenacity

最基本的重試

無條件重試,重試之間無間隔

from tenacity import retry

@retry

def test_retry():

print(“等待重試,重試無間隔執行。。。”)

raise Exception

test_retry()

無條件重試,但是在重試之前要等待 2 秒

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(2))

def test_retry():

print(“等待重試。。。”)

raise Exception

test_retry()

設定停止基本條件

只重試7 次

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7))

def test_retry():

print(“等待重試。。。”)

raise Exception

test_retry()

重試 10 秒後不再重試

from tenacity import retry, stop_after_delay

@retry(stop=stop_after_delay(10))

def test_retry():

print(“等待重試。。。”)

raise Exception

test_retry()

或者上面兩個條件滿足一個就結束重試

from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(10) | stop_after_attempt(7)))

def test_retry():

print(“等待重試。。。”)

raise Exception

test_retry()

設定何時進行重試

在出現特定錯誤/異常(比如請求超時)的情況下,再進行重試

from requests import exceptions

from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type(exceptions。Timeout))

def test_retry():

print(“等待重試。。。”)

raise exceptions。Timeout

test_retry()

在滿足自定義條件時,再進行重試。

如下示例,當

test_retry

函式返回值為 False 時,再進行重試

from tenacity import retry, stop_after_attempt, retry_if_result

def is_false(value):

return value is False

@retry(stop=stop_after_attempt(3),

retry=retry_if_result(is_false))

def test_retry():

return False

test_retry()

重試後錯誤重新丟擲

當出現異常後,tenacity 會進行重試,若重試後還是失敗,預設情況下,往上丟擲的異常會變成 RetryError,而不是最根本的原因。

因此可以加一個引數(

reraise=True

),使得當重試失敗後,往外丟擲的異常還是原來的那個。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7), reraise=True)

def test_retry():

print(“等待重試。。。”)

raise Exception

test_retry()

設定回撥函式

當最後一次重試失敗後,可以執行一個回撥函式

from tenacity import *

def return_last_value(retry_state):

print(“執行回撥函式”)

return retry_state。outcome。result() # 表示返回原函式的返回值

def is_false(value):

return value is False

@retry(stop=stop_after_attempt(3),

retry_error_callback=return_last_value,

retry=retry_if_result(is_false))

def test_retry():

print(“等待重試中。。。”)

return False

print(test_retry())

輸出如下

等待重試中。。。

等待重試中。。。

等待重試中。。。

執行回撥函式

False

標簽: retry  重試  stop  After  tenacity