demo如下:
from retrying import retry def retry_error(exception): return isinstance(exception, NameError) # return isinstance(exception, Exception) @retry(retry_on_exception=retry_error, stop_max_attempt_number=3) def demo_(): print('重試') print(a) if __name__ == '__main__': demo_()
運行如圖:
定義了一個demo_函數,里面打印的變量a,沒有被賦值,會導致報“NameError: name 'a' is not defined”錯誤,但是觀察上述截圖,發現代碼運行了3次以后才拋出異常。這就借用了retrying里面的失敗重試機制,定義了retry_error函數去捕獲NameError異常,其實也可以偷懶直接去捕獲Exception。用retry_on_result去
傳入一個函數對象,stop_max_attempt_number控制次數。
以上參考博客:https://www.cnblogs.com/du-jun/p/12256281.html
自己實現了一個失敗重試裝飾器的demo:
# def retry(**kw): # def wrapper(func): # def _wrapper(*args, **kwargs): # raise_ex = 0 # for _ in range(kw['retry_num']): # print(f"第{_}次操作") # try: # return func(*args, **kwargs) # except Exception as e: # raise_ex += 1 # if raise_ex == kw['retry_num']: # raise e # # return _wrapper # # return wrapper # # # @retry(retry_num=3) # def demo(): # print('a') # print(a) # # raise NameError # # # if __name__ == '__main__': # demo() def retry(loop_num=2): def wrapper(func): def _wrapper(*args, **kwargs): raise_ex = 0 for _ in range(loop_num): print(f"第{_}次操作") try: return func(*args, **kwargs) except Exception as e: raise_ex += 1 if raise_ex == loop_num: raise e return _wrapper return wrapper @retry(loop_num=3) def demo(): print('a') print(a) # raise NameError if __name__ == '__main__': demo()
效果如圖:
增加等待時間:
import time def retry(loop_num=2, wait_time=1): """ :param loop_num: 循環次數,默認2次 :param wait_time: 等待時間,默認1s :return: """ def wrapper(func): def _wrapper(*args, **kwargs): raise_ex = 0 for i in range(1, loop_num + 1): print(f"第{i}次操作") try: func(*args, **kwargs) except Exception as e: time.sleep(wait_time) raise_ex += 1 if raise_ex == loop_num: raise e return _wrapper return wrapper @retry() def demo_(): print('開始重試') print(a) if __name__ == '__main__': demo_()