1. 程式人生 > 其它 >一個python實現重試機制的簡要實踐

一個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即可(所以如果有遇到和我類似情況的,把那些類似只需獲取一次資料的方法放到外面,避免重複請求介面引發異常)