一個python實現重試機制的簡要實踐

最近在寫接口測試腳本時，遇到如下一個測試場景

1、A系統會創建一條數據，創建成功后會把數據推到B系統；

2、由於是兩個系統之間通信，數據不會立刻從A系統同步到B系統，中間有一個短暫的時間差；

我要調試的接口有2個，一是在A系統調用一個接口，生成數據；二是在B系統調用另一個接口處理數據。

實際操作后，發現一個問題：由於調用完A接口后，會立刻調用B接口，從代碼層面看，這個時間差很短，雖然A系統已經把數據生成了，但在這么短時間內還沒推送到B系統，導致調用B接口時，查不到這條數據，就會報錯

第一個解決方案

開始想到的解決方案是使用time.sleep()，當調用A接口后，等待一段時間，如 time.sleep(5)，死等5s，然后再調用B接口

因為等待5s后，數據一般能夠從A系統推送到B系統

當然如果5s后還沒有同步到B系統，調用B接口時仍然會報錯，所以這並不是一個很好的解決方案

第二個解決方案

互聯網沖浪一番后發現了python有一個庫可以實現重試機制：tenacity

下面是找到的一些參考博客，可以看一下基礎用法：https://www.cnblogs.com/wuzhibinsuib/p/13443622.html

接下來說一下自己的實驗結果以及理解

它的簡單用法是給需要重試的代碼加上@retry修飾器，代碼拋出異常會被裝飾器捕獲並進行重試

這里的關鍵是捕獲到到代碼拋出的異常

 

例1【如果報錯會一直重試】

@retry
def test_retry1():
    print("等待重試.....")
    raise Exception  # 通過raise直接返回一個錯誤

@retry
def test_retry2():
    print("等待重試.....")
    return "hello" + 1  # 人為制造一個錯誤，這里我是把字符串和整數相加，因為類型不同，肯定會報錯，所以會觸發重試

上述2段代碼運行后會一直打印“等待重試”，直至手工停止運行

 

例2【設置最大重試次數stop_after_attempt】

@retry(stop=stop_after_attempt(5))
def test_retry():
    print("等待重試.....")
    return "hello" + 1 

用 stop 接收 stop_after_attempt，當重試指定次數時，結束重試，如下重試了5次

 

例3【設置最大重時間，如果失敗，則重試，一直重試5s】 

@retry(stop=stop_after_delay(5))
def test_retry():
    print("等待重試.....")
    return "hello" + 1 

 

例4【出現特定錯誤后重試】

@retry(retry=retry_if_exception_type(TypeError))
def test_retry1():
    print("等待重試.....")
    return "hello" + 1  # 捕獲類型錯誤，當出現類型錯誤時重試

@retry(retry=retry_if_exception_type(SyntaxError))
def test_retry2():
    print("等待重試.....")
    raise SyntaxError  # 捕獲語法錯誤，當出現語法錯誤時重試

 

例5【滿足自定義的條件后重試】

# 首先定義了一個函數symbol，它的作用是判斷傳入的值是否為None；它返回一個布爾值，如果結果value=None，則返回true，否則返回False
def symbol(value):
    return value is None


# 裝飾器中retry=retry_if_result(symbol)，表示把test_retry函數的結果傳入symbol，判斷test_retry的結果是否為None，
# 如果=None，就進行重試(retry),如果不等於None，就結束並返回函數值（所以達成重試的條件是test_retry的結果是否為條件函數定義的結果）
@retry(stop=stop_after_attempt(3), retry=retry_if_result(symbol), reraise=True)
def test_retry():
    print("等待重試.....")
    return None

symbol()函數是定義的條件函數，test_retry()函數是希望重試的函數，它倆通過裝飾器中的retry_if_result()來關聯，具體含義可以看下上述代碼的注釋

---

接下來開始處理我的接口測試腳本，用到是上面例5的自定義條件重試

首先處理需要重試的方法，我規定了當這個方法沒有接收到推送過來的數據時，返回None

    def seal_regist(code):

        seal_data = self.get_seal_data(code)

        try:
            if seal_data["data"]["list"]:

                r = requests.post(url, json=payload, headers=headers)
                return r.json()

            else:
                print("列表中沒有這條待用印數據{}，請檢查系統～".format(code))
                return None

        except Exception as e:
            raise e

定義一個條件函數

def test_retry(value):
    """重試條件函數"""
    return value is None

給 seal_regist()函數加上retry裝飾器

@retry(stop=stop_after_delay(10), retry=retry_if_result(test_retry))
    def seal_regist(code):
　　　　　　.....
　　　　　　.....

如果 seal_regist()返回None則重試，最大重試時間為10s

ps.因為重試函數中需要用到登陸cookie，之前是把登陸獲取cookie的方法寫到里面的，但是如果加上重試機制的話，當開始重試時會一直重新登錄獲取cookie，提示登陸頻繁並導致登陸接口調用失敗，所以為了避免這種情況，我把獲取登陸cookie的方法放到了外面，這樣無論重試幾次都用開始獲取到的一個cookie即可（所以如果有遇到和我類似情況的，把那些類似只需獲取一次數據的方法放到外面，避免重復請求接口引發異常）