一個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 deftest_retry1(): print("等待重試.....") raise Exception # 通過raise直接返回一個錯誤 @retry def test_retry2(): print("等待重試.....") return "hello" + 1 # 人為製造一個錯誤,這裡我是把字串和整數相加,因為型別不同,肯定會報錯,所以會觸發重試
上述2段程式碼執行後會一直列印“等待重試”,直至手工停止執行
例2【設定最大重試次數stop_after_attempt】
@retry(stop=stop_after_attempt(5)) deftest_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即可(所以如果有遇到和我類似情況的,把那些類似只需獲取一次資料的方法放到外面,避免重複請求介面引發異常)