tenacity.retry Python中一個專門用來重試的庫

tenacity.retry Python中一個專門用來重試的庫

一、背景：

很多時候，我們都喜歡為代碼加入retry功能，比如oauth驗證，有時候網路不太靈，我們希望多試幾次，

這些retry應用的場景看起來不同，其實又很類似，都是判斷代碼是否正常運行，如果不是則重新開始，

那么，有沒有一種通用的辦法來實作呢？

二、簡介：

Tenacity1是一個通用的retry庫，簡化為任何任務加入重試的功能，

它還包含如下特性:

• 通用的裝飾器API
• 可以設定重試停止的條件（比如設定嘗試次數）
• 可以設定重試間的等待時間（比如在嘗試之間使用冪數級增長的wait等待）
• 自定義在哪些Exception進行重試
• 自定義在哪些回傳值的情況進行重試
• 協程的重試

三、用法

1、引數講解

① 不要引數會一直重試，也沒有時間等待
② stop:重試停止方式，
③ wait：每次重試間隔時間，
④ retry：什么樣的情況下才重試（可以指定特定的錯誤）
⑤ before：重試前記日志
⑥ after：重試后記日志

2、實列

① 重試次數與等待時間

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

我自己代碼中用到了: OperationalError\InternalError這兩種錯誤下才會重試，重試3次，每集間隔20s

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(20), retry=(retry_if_exception_type(OperationalError)|retry_if_exception_type(InternalError)),reraise=True)
    def sql_to_many(self, *args):
        """
        批量修改和存盤 支支持update\insert
        :param args: args[0]:資料庫， args[1]:sql， args[2]:data [tuple,tuple]
        :return:
        """
        self.connect_(args[0])
        self.executemany_sql(args[1], args[2])
        self.close_()

以上就是 Python retry庫的介紹了，用起來真的很方便哦~