1 Tenacity描述

今天 給大家介紹一個(gè)Python 重試庫(kù)，Tenacity 這個(gè)庫(kù) 是我 這些年 使用的一個(gè)非常好的庫(kù)，幾乎滿足了我所有的重試需求，非常棒的一個(gè)庫(kù)。

在我們?nèi)粘ｉ_(kāi)發(fā)中，我們經(jīng)常會(huì)遇到 網(wǎng)絡(luò)問(wèn)題 或者其他問(wèn)題，導(dǎo)致 API請(qǐng)求失敗，或者超時(shí)等出現(xiàn)， 這個(gè)出現(xiàn)是偶發(fā)性，不確定性的，我們?nèi)绾伪ＷC系統(tǒng)的功能的正確性呢？ 第一種 顯示 告訴 用戶請(qǐng)求失敗。 第二種 我們可以通過(guò)重試 幾次 看看會(huì)不會(huì)成功，如果 能成功 就可以了。

因此在真實(shí)的項(xiàng)目開(kāi)發(fā)中重試的庫(kù)，也顯得 尤其重要， 在工作中 我也有嘗試寫過(guò)一些 重試的裝飾器函數(shù)，總感覺(jué) 差點(diǎn)意思，要不就是 考慮不全面，要么就是 實(shí)現(xiàn)太簡(jiǎn)單 ，不能針對(duì)具體的異常進(jìn)行重試，不能設(shè)置一些條件重試等。 以及重試 還是失敗了怎么處理？ 等等一系列的問(wèn)題。

直到后來(lái)發(fā)現(xiàn)這個(gè) 寶藏庫(kù) tenacity ，幾乎解決了我所有的痛點(diǎn)。

一起來(lái)了解這個(gè)庫(kù)的基本用法

2 如果發(fā)生異常就重試

這個(gè)就是 如果發(fā)生了異常就一直重試下去：

from tenacity import retry, stop_after_attempt
@retry()
def never_gonna_give_you_up():
    print("Retry forever ignoring Exceptions, don't wait between retries")
    raise Exception("Unknown Error")

3 設(shè)置停止重試的條件

設(shè)置重試的最大次數(shù)

這里設(shè)置最多重試5次 ，失敗后 就不在進(jìn)行重試了。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(5))
def never_gonna_give_you_up():
    print("Retry forever ignoring Exceptions, don't wait between retries")
    raise Exception("Unknown Error")

if __name__ == '__main__':
    never_gonna_give_you_up()

還可以設(shè)置stop 時(shí)間

stop = stop_after_delay

Stop when the time from the first attempt >= limit 從第一次失敗 嘗試 到 多久后 停止嘗試。

默認(rèn) stop_after_delay 單位是秒

import time
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
    print("10秒后停止重試")
    time.sleep(5)
    raise Exception("Unknown Error")
if __name__ == '__main__':
    stop_after_10_s()

停止重試條件 進(jìn)行組合

from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
    print("10秒后或者5次重試后停止重試")
    raise Exception("Unknown Error")
if __name__ == '__main__':
    stop_after_10_s_or_5_retries()

4 設(shè)置重試的間隔

有的時(shí)候 重試 太頻繁了，也不太好，可能因?yàn)?網(wǎng)絡(luò)問(wèn)題，過(guò)一段時(shí)間才能恢復(fù)好，

wait_fixed(seconds=10) 等幾秒后重試， 然后 可以組合 stop 在一起使用。

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3),wait=wait_fixed(3))
def never_gonna_give_you_up():
    print("Retry max attempt:3, wait_fixed:3")
    raise Exception("Unknown Error")
if __name__ == '__main__':
    never_gonna_give_you_up()

5 重試觸發(fā)的條件

針對(duì)具體的異常進(jìn)行重試

對(duì)具體的異常進(jìn)行重試 ,而不是所有異常進(jìn)行重試. 有的時(shí)候 我們?cè)趯?業(yè)務(wù)代碼的時(shí)候，可能會(huì)定義一些 自定義的異常，我們有時(shí)候 觸發(fā)特定的異常 進(jìn)行重試，而不是說(shuō) 發(fā)生了異常 就進(jìn)行 重試。 tenacity 這個(gè)庫(kù) 早就想好了這點(diǎn)。

我自定義了一個(gè)異常 NotExistResourceErr ， 觸發(fā)這個(gè)異常我才進(jìn)行重試，重試最多三次。

# -*- coding: utf-8 -*-
from typing import Any

from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type
class NotExistResourceErr(Exception):
    """
    定義一個(gè)自定義的異常
    """
    code = str(400)
    detail = "Parameter Error"
    def __init__(self, code: str = None, detail: Any = None):
        self.code = code
        self.detail = detail

class Source:
    def __init__(self, id):
        self.id = id

@retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3))
def might_exist_error(task_id: int = 0):
    print('might_exist_error begin ...')
    if task_id < 0:
        raise NotExistResourceErr()
    return Source(task_id)

if __name__ == '__main__':
    might_exist_error(-10)

retry_if_exception_type 這里就告訴 tenacity 在發(fā)生什么類型，才進(jìn)行重試操作。 設(shè)置 stop_after_attempt(3) 重試3 次就停止了。

運(yùn)行一下結(jié)果如下： 可以看出 重試了三次，第三次后 拋出了異常

如果在里面 這樣 拋出一個(gè)異常的話，此時(shí) tenacity 不會(huì)進(jìn)行重試操作，而是 程序直接報(bào)錯(cuò)。

@retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3))
def might_exist_error(task_id: int = 0):
    print('might_exist_error begin ...')
    if task_id < 0:
        raise Exception("Unknown Error")

    return Source(task_id)

如果這樣的話， 不會(huì)進(jìn)行重試操作 ，而是直接拋出了異常。

針對(duì)函數(shù)的返回結(jié)果進(jìn)行重試

有的時(shí)候 可能因?yàn)?某些原因 正常情況下，應(yīng)該可以獲取到值，但是返回結(jié)果為None, 異常情況 在 函數(shù)內(nèi)部進(jìn)行捕獲處理了，這個(gè)時(shí)候 也想進(jìn)行 重試 怎么辦呢？

retry_if_result 這里可以傳入一個(gè)函數(shù)，這個(gè)函數(shù)接收 might_return_none 返回值， 然后 這個(gè)函數(shù) 返回 True 就會(huì) 觸發(fā)重試操作。

from tenacity import retry, stop_after_attempt, retry_if_result

def is_none(value):
    """Return True if value is None"""
    return value is None

@retry(retry=retry_if_result(is_none), stop=stop_after_attempt(3))
def might_return_none():
    print("返回 None 則重試")
    return None
if __name__ == '__main__':
    might_return_none()

類似與此相反的 retry_if_not_result 當(dāng)結(jié)果 不符合預(yù)期 則進(jìn)行重試操作。

6 定義重試失敗回調(diào)函數(shù)

對(duì)于重要的業(yè)務(wù)的話， 如果重試幾次后， 發(fā)現(xiàn)仍然是失敗的狀態(tài)，此時(shí) 我們可以 設(shè)置 一個(gè)回調(diào)函數(shù)， 比如 在回調(diào)函數(shù)中 發(fā)一個(gè)郵件 通知 相關(guān) 的人員，手動(dòng)處理異常，檢查問(wèn)題等。

retry_error_callback 這里可以傳入一個(gè)回調(diào) 函數(shù) ， 回函函數(shù)中有一個(gè)參數(shù) retry_state 最后一次 函數(shù)的返回的狀態(tài)。

在回調(diào)函數(shù)中 可以做你想做的任何處理，發(fā)郵件 或者其他的操作。

from tenacity import stop_after_attempt, retry, retry_if_result
def send_email(msg):
    print(msg)
# 回調(diào)函數(shù)
def callback(retry_state):
    """return the result of the last call attempt"""
    # print(f"call function ... {retry_state}, {type(retry_state)}")
    send_email('eventually_return_false eventually failed')
    return retry_state.outcome.result()

def is_false(value):
    """Return True if value is False"""
    return value is False

# will return False after trying 3 times to get a different result
@retry(stop=stop_after_attempt(3),
       retry_error_callback=callback,
       retry=retry_if_result(is_false))
def eventually_return_false():
    print("failed ...")
    return False

if __name__ == '__main__':
    eventually_return_false()
    pass

7 錯(cuò)誤處理

有的時(shí)候 我們可能重試后失敗了，想獲取原來(lái) 程序代碼中拋出 的異常。 我們可以使用 reraise 參數(shù)來(lái) 拋出異常。是一個(gè)bool 變量，

設(shè)置為True 會(huì)拋出原來(lái)的異常， 如果設(shè)置為False 則不會(huì)拋出原來(lái)的異常 ,會(huì)拋出 tenacity.RetryError

from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type, retry_if_result
class NotExistResourceErr(Exception):
    """
    定義一個(gè)自定義的異常
    """
    code = str(400)

    detail = "Parameter Error"

    def __init__(self, code: str = None, detail: Any = None):
        self.code = code or self.code
        self.detail = detail or self.detail

@retry(reraise=True , stop=stop_after_attempt(3))
def raise_my_exception():
    raise NotExistResourceErr(detail="Fail",code='0000')

try:
    raise_my_exception()
except NotExistResourceErr as e :
    print(f'handle exception detail:{e.detail} code:{e.code}')

設(shè)置為 reraise=False 整個(gè)程序 就會(huì)掛掉，如下面的報(bào)錯(cuò)：

Traceback (most recent call last):
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 407, in __call__
    result = fn(*args, **kwargs)
  File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 42, in raise_my_exception
    raise NotExistResourceErr("Fail")
__main__.NotExistResourceErr: Fail

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 45, in <module>
    raise_my_exception()
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 324, in wrapped_f
    return self(f, *args, **kw)
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 404, in __call__
    do = self.iter(retry_state=retry_state)
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 361, in iter
    raise retry_exc from fut.exception()
tenacity.RetryError: RetryError[<Future at 0x7f990dddb4c0 state=finished raised NotExistResourceErr>]

8 推薦看下 作者的對(duì)這個(gè)庫(kù)介紹

做了一些翻譯，感興趣的同學(xué) 可以關(guān)注一下。Python never gives up: the tenacity library 原文

Python never gives up: the tenacity library

A couple of years ago, I wrote about the Python retrying library. This library was designed to retry the execution of a task when a failure occurred.

幾年前，我寫過(guò)關(guān)于Python重試庫(kù)的文章。這個(gè)庫(kù)被設(shè)計(jì)用來(lái)在一個(gè)任務(wù)發(fā)生故障時(shí)重試執(zhí)行。

I started to spread usage of this library in various projects, such as Gnocchi, these last years. Unfortunately, it started to get very hard to contribute and send patches to the upstream retrying project. I spent several months trying to work with the original author. But after a while, I had to conclude that I would be unable to fix bugs and enhance it at the pace I would like to. Therefore, I had to take a difficult decision and decided to fork the library.

在過(guò)去的幾年里，我開(kāi)始在各種項(xiàng)目中推廣使用這個(gè)庫(kù)，比如Gnocchi。不幸的是，向上游的retrying項(xiàng)目貢獻(xiàn)和發(fā)送補(bǔ)丁開(kāi)始變得非常困難。我花了幾個(gè)月的時(shí)間試圖與原作者合作。但一段時(shí)間后，我不得不得出結(jié)論，我將無(wú)法以我想要的速度來(lái)修復(fù)錯(cuò)誤和增強(qiáng)它。因此，我不得不做出一個(gè)艱難的決定，決定fork這個(gè)庫(kù)。

Here comes tenacity

I picked a new name and rewrote parts of the API of retrying that were not working correctly or were too complicated. I also fixed bugs with the help of Joshua, and named this new library tenacity. It works in the same manner as retrying does, except that it is written in a more functional way and offers some nifty new features.

tenacity 就這樣誕生了，

我選了一個(gè)新的名字，并重寫了重試的API中不能正確工作或過(guò)于復(fù)雜的部分。我還在Joshua的幫助下修復(fù)了一些錯(cuò)誤，并將這個(gè)新的庫(kù)命名為tenacity。它的工作方式與重試相同，只是它是以一種更實(shí)用的方式編寫的，并提供一些有趣的新功能。

nifty /?n?ft?/

niftier 比較級(jí)

niftiest 最高級(jí)

ADJ If you describe something as nifty, you think it is neat and pleasing or cleverly done. 整潔而惹人喜愛(ài)的; 完成得精巧的

• Bridgeport was a pretty nifty place.

布里奇波特曾是個(gè)相當(dāng)整潔、惹人喜愛(ài)的地方。

• It was a nifty arrangement, a perfect partnership.

這既是個(gè)絕佳的安排，又是個(gè)完美的合作。

Basic usage

The basic usage is to use it as a decorator:

import tenacity

@tenacity.retry
def do_something_and_retry_on_any_exception():
    pass

This will make the function do_something_and_retry_on_any_exception be called over and over again until it stops raising an exception. It would have been hard to design anything simpler. Obviously, this is a pretty rare case, as one usually wants to e.g. wait some time between retries. For that, tenacity offers a large panel of waiting methods:

這將使函數(shù)do_something_and_retry_on_any_exception被反復(fù)調(diào)用，直到它停止引發(fā)異常。很難再設(shè)計(jì)出更簡(jiǎn)單的東西了。顯然，這是一個(gè)相當(dāng)罕見(jiàn)的情況，因?yàn)槿藗兺ǔＯＭ谥卦囍g等待一段時(shí)間。為此，tenacity提供了大量的等待方法。

import tenacity

@tenacity.retry(wait=tenacity.wait_fixed(1))
def do_something_and_retry():
    do_something()

Or a simple exponential back-off method can be used instead:

或者可以用一個(gè)簡(jiǎn)單的指數(shù)回退法來(lái)代替。

import tenacity

@tenacity.retry(wait=tenacity.wait_exponential())
def do_something_and_retry():
    do_something()

Combination

組合

What is especially interesting with tenacity, is that you can easily combine several methods. For example, you can combine tenacity.wait.wait_random with tenacity.wait.wait_fixed to wait a number of seconds defined in an interval:

tenacity特別有趣的是，你可以很容易地結(jié)合幾種方法。例如，你可以把 tenacity.wait.wait_random 和 tenacity.wait.wait_fixed 結(jié)合起來(lái)，等待在一個(gè)時(shí)間間隔內(nèi)定義的若干秒。

import tenacity

@tenacity.retry(wait=tenacity.wait_fixed(10) + wait.wait_random(0, 3))
def do_something_and_retry():
    do_something()

This will make the function being retried wait randomly between 10 and 13 seconds before trying again.

這將使被重試的函數(shù)在再次嘗試之前隨機(jī)地等待10到13秒。

tenacity offers more customization, such as retrying on some exceptions only. You can retry every second to execute the function only if the exception raised by do_something is an instance of IOError, e.g. a network communication error.

tenacity提供了更多的自定義功能，比如只在某些異常情況下重試。你可以在do_something引發(fā)的異常是IOError的實(shí)例時(shí)，例如網(wǎng)絡(luò)通信錯(cuò)誤，才每秒重試執(zhí)行函數(shù)。

import tenacity

@tenacity.retry(wait=tenacity.wait_fixed(1),
                retry=tenacity.retry_if_exception_type(IOError))
                
def do_something_and_retry():
    do_something()

You can combine several condition easily by using the | or & binary operators. They are used to make the code retry if an IOError exception is raised, or if no result is returned. Also, a stop condition is added with the stop keyword arguments. It allows to specify a condition unrelated to the function result of exception to stop, such as a number of attemps or a delay.

你可以通過(guò)使用|或&二進(jìn)制操作符輕松地組合幾個(gè)條件。它們被用來(lái)在引發(fā)IOError異常或沒(méi)有返回結(jié)果時(shí)使代碼重試。此外，還可以用stop關(guān)鍵字參數(shù)添加一個(gè)停止條件。它允許指定一個(gè)與要停止的異常的函數(shù)結(jié)果無(wú)關(guān)的條件，例如嘗試的次數(shù)或延遲。

import tenacity

@tenacity.retry(wait=tenacity.wait_fixed(1),
                stop=tenacity.stop_after_delay(60),
                retry=(tenacity.retry_if_exception_type(IOError) |
                       tenacity.retry_if_result(lambda result: result == None))
def do_something_and_retry():
    do_something()

The functional approach of tenacity makes it easy and clean to combine a lot of condition for various use cases with simple binary operators.

tenacity的函數(shù)式方法使得它可以很容易和干凈地用簡(jiǎn)單的二進(jìn)制運(yùn)算符為各種使用情況組合大量的條件。

Standalone usage

獨(dú)立使用

tenacity can also be used without decorator by using the object Retrying, which implements its main behavior and using its call method. This allows to call any function with different retry conditions, or to retry any piece of code that do not use the decorator at all – like code from an external library.

tenacity也可以在沒(méi)有裝飾器的情況下使用，通過(guò)使用Retrying對(duì)象，實(shí)現(xiàn)其主要行為并使用其調(diào)用方法。這允許調(diào)用任何具有不同重試條件的函數(shù)，或重試任何完全不使用裝飾器的代碼–如來(lái)自外部庫(kù)的代碼。

import tenacity

r = tenacity.Retrying(
    wait=tenacity.wait_fixed(1),
    retry=tenacity.retry_if_exception_type(IOError))

r.call(do_something)

This also allows you to re-use that object without creating one new one each time, saving some memory!

這也允許你重復(fù)使用該對(duì)象，而不需要每次都創(chuàng)建一個(gè)新的對(duì)象，從而節(jié)省一些內(nèi)存。

I hope you’ll like it and will find it of some use. Feel free to fork it, report bugs, or ask for new features on its GitHub!

我希望你會(huì)喜歡它，并會(huì)發(fā)現(xiàn)它有一些用處。歡迎在GitHub上分享它，報(bào)告錯(cuò)誤，或要求提供新的功能。

If you want to learn more about retrying strategy and how to handle failure, there’s even more in Scaling Python. Check it out!

到此這篇關(guān)于Python重試庫(kù) Tenacity詳解(推薦)的文章就介紹到這了,更多相關(guān)Python Tenacity 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論