Python中最強(qiáng)大的錯(cuò)誤重試庫
我們?cè)诰帉懗绦蛴绕涫桥c網(wǎng)絡(luò)請(qǐng)求相關(guān)的程序,如調(diào)用web接口、運(yùn)行網(wǎng)絡(luò)爬蟲等任務(wù)時(shí),經(jīng)常會(huì)遇到一些偶然發(fā)生的請(qǐng)求失敗的狀況,這種時(shí)候如果我們僅僅簡單的捕捉錯(cuò)誤然后跳過對(duì)應(yīng)任務(wù),肯定是不嚴(yán)謹(jǐn)?shù)?,尤其是在網(wǎng)絡(luò)爬蟲中,會(huì)存在損失有價(jià)值數(shù)據(jù)的風(fēng)險(xiǎn)。
這類情況下我們就很有必要為我們的程序邏輯添加一些「錯(cuò)誤重試」的策略,費(fèi)老師我在幾年前寫過文章介紹過Python中的tenacity庫,但它功能較為單一,只能應(yīng)對(duì)基本的需求。
而今天我要給大家介紹的tenacity庫,可能是目前Python生態(tài)中最好用的錯(cuò)誤重試庫,下面就讓我們一睹其主要功能吧!
tenacity中的常用功能
作為一個(gè)第三方Python庫,我們可以使用pip install tenacity對(duì)其進(jìn)行安裝,安裝完成后,下面我們來學(xué)習(xí)一下tenacity的主要使用方法和特性:
1. tenacity的基礎(chǔ)使用
tenacity的錯(cuò)誤重試核心功能由其retry裝飾器來實(shí)現(xiàn),默認(rèn)不給retry裝飾器傳參數(shù)時(shí),它會(huì)在其所裝飾的函數(shù)運(yùn)行過程拋出錯(cuò)誤時(shí)不停地重試下去,譬如下面這個(gè)簡單的例子:
import random
from tenacity import retry
@retry
def demo_func1():
a = random.random()
print(a)
if a >= 0.1:
raise Exception
demo_func1()
可以看到,我們的函數(shù)體內(nèi)每次生成0到1之間的隨機(jī)數(shù),當(dāng)這個(gè)隨機(jī)數(shù)不超過0.1時(shí)才會(huì)停止拋出錯(cuò)誤,否則則會(huì)被tenacity捕捉到每次的錯(cuò)誤拋出行為并立即重試。
2. 設(shè)置最大重試次數(shù)
有些時(shí)候我們對(duì)某段函數(shù)邏輯錯(cuò)誤重試的忍耐是有限度的,譬如當(dāng)我們調(diào)用某個(gè)網(wǎng)絡(luò)接口時(shí),如果連續(xù)n次都執(zhí)行失敗,我們可能就會(huì)認(rèn)為這個(gè)任務(wù)本身就存在缺陷,不是通過重試就能有朝一日正常的。
這種時(shí)候我們可以利用tenacity中的stop_after_attempt函數(shù),作為retry()中的stop參數(shù)傳入,從而為我們“無盡”的錯(cuò)誤重試過程添加一個(gè)終點(diǎn),其中stop_after_attempt()接受一個(gè)整數(shù)輸入作為「最大重試」的次數(shù):
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def demo_func2():
print('函數(shù)執(zhí)行')
raise Exception
demo_func2()
可以看到,我們的函數(shù)在限制了最大重試次數(shù)后,經(jīng)過3次重試,在第4次繼續(xù)執(zhí)行依然拋出錯(cuò)誤后,正式地拋出了函數(shù)中對(duì)應(yīng)的Exception錯(cuò)誤結(jié)束了重試過程。
3. 設(shè)置重試最大超時(shí)時(shí)長
我們除了像上一小節(jié)中那樣設(shè)置最大錯(cuò)誤重試的次數(shù)之外,tenacity還為我們提供了stop_after_delay()函數(shù)來設(shè)置整個(gè)重試過程的最大耗時(shí),超出這個(gè)時(shí)長也會(huì)結(jié)束重試過程:
import time
from tenacity import retry, stop_after_delay
# 設(shè)置重試最大超時(shí)時(shí)長為5秒
@retry(stop=stop_after_delay(5))
def demo_func3():
time.sleep(1)
print(f'已過去 {time.time() - start_time} 秒')
raise Exception
# 記錄開始時(shí)間
start_time = time.time()
demo_func3()
4. 組合重試停止條件
如果我們的任務(wù)同時(shí)需要添加最大重試次數(shù)以及最大超時(shí)時(shí)長限制,在tenacity中僅需要用|運(yùn)算符組合不同的限制條件再傳入retry()的stop參數(shù)即可,譬如下面的例子,當(dāng)我們的函數(shù)執(zhí)行重試超過3秒或次數(shù)大于5次時(shí)均可以結(jié)束重試:
import time
import random
from tenacity import retry, stop_after_delay, stop_after_attempt
@retry(stop=(stop_after_delay(3) | stop_after_attempt(5)))
def demo_func4():
time.sleep(random.random())
print(f'已過去 {time.time() - start_time} 秒')
raise Exception
# 記錄開始時(shí)間
start_time = time.time()
demo_func4()
可以看到,在上面的演示中,先達(dá)到了“最大重試5次”的限制從而結(jié)束了重試過程。
5. 設(shè)置相鄰重試之間的時(shí)間間隔
有些情況下我們并不希望每一次重試拋出錯(cuò)誤后,立即開始下一次的重試,譬如爬蟲任務(wù)中為了更好地偽裝我們的程序,tenacity中提供了一系列非常實(shí)用的函數(shù),配合retry()的wait參數(shù),幫助我們妥善處理相鄰重試之間的時(shí)間間隔,其中較為實(shí)用的主要有以下兩種方式:
(1) 設(shè)置固定時(shí)間間隔
我們通過使用tenacity中的wait_fixed()可以為相鄰重試之間設(shè)置固定的等待間隔秒數(shù),就像下面的簡單示例那樣:
import time
from tenacity import retry, wait_fixed, stop_after_attempt
# 設(shè)置重試等待間隔為1秒
@retry(wait=wait_fixed(1), stop=stop_after_attempt(3))
def demo_func5():
print(f'已過去 {time.time() - start_time} 秒')
raise Exception
# 記錄開始時(shí)間
start_time = time.time()
demo_func5()
(2) 設(shè)置隨機(jī)時(shí)間間隔
除了設(shè)置固定的時(shí)間間隔外,tenacity還可以通過wait_random()幫助我們?yōu)橄噜徶卦囋O(shè)置均勻分布隨機(jī)數(shù),只需要設(shè)置好均勻分布的范圍即可:
import time
from tenacity import retry, wait_random, stop_after_attempt
# 設(shè)置重試等待間隔為1到3之間的隨機(jī)數(shù)
@retry(wait=wait_random(min=1, max=3), stop=stop_after_attempt(5))
def demo_func6():
print(f'已過去 {time.time() - start_time} 秒')
raise Exception
# 記錄開始時(shí)間
start_time = time.time()
demo_func6()
可以觀察到,每一次重試后的等待時(shí)長都是隨機(jī)的~
6. 自定義是否觸發(fā)重試
tenacity中retry()的默認(rèn)策略是當(dāng)其所裝飾的函數(shù)執(zhí)行過程“拋出任何錯(cuò)誤”時(shí)即進(jìn)行重試,但有些情況下我們需要的可能是對(duì)特定錯(cuò)誤類型的捕捉/忽略,亦或是對(duì)異常計(jì)算結(jié)果的捕捉。
tenacity中同樣內(nèi)置了相關(guān)的實(shí)用功能:
(1) 捕捉或忽略特定的錯(cuò)誤類型
使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type(),配合retry()的retry參數(shù),我們可以對(duì)特定的錯(cuò)誤類型進(jìn)行捕捉或忽略:
from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type
@retry(retry=retry_if_exception_type(FileExistsError))
def demo_func7():
raise TimeoutError
@retry(retry=retry_if_not_exception_type(FileNotFoundError))
def demo_func8():
raise FileNotFoundError
(2) 自定義函數(shù)結(jié)果條件判斷函數(shù)
我們可以編寫額外的條件判斷函數(shù),配合tenacity中的retry_if_result(),實(shí)現(xiàn)對(duì)函數(shù)的返回結(jié)果進(jìn)行自定義條件判斷,返回True時(shí)才會(huì)觸發(fā)重試操作:
import random
from tenacity import retry, retry_if_result
@retry(retry=retry_if_result(lambda x: x >= 0.1))
def demo_func9():
a = random.random()
print(a)
return a
# 記錄開始時(shí)間
demo_func9()
7. 對(duì)函數(shù)的錯(cuò)誤重試情況進(jìn)行統(tǒng)計(jì)
被tenacity的retry()裝飾的函數(shù),我們可以打印其retry.statistics屬性查看其歷經(jīng)的錯(cuò)誤重試統(tǒng)計(jì)記錄結(jié)果,譬如這里我們對(duì)前面執(zhí)行過的示例函數(shù)demo_func9()的統(tǒng)計(jì)結(jié)果進(jìn)行打?。?/p>
demo_func9.retry.statistics
除了上述的功能之外,tenacity還具有很多特殊的特性,可以結(jié)合logging模塊、異步函數(shù)、協(xié)程等其他Python功能實(shí)現(xiàn)更高級(jí)的功能,感興趣的朋友可以前往https://github.com/jd/tenacity了解更多。