1 Tenacity描述

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

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

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

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

一起來了解這個庫的基本用法

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

這個就是 如果發(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 時間

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è)置重試的間隔

有的時候 重試 太頻繁了，也不太好，可能因?yàn)?網(wǎng)絡(luò)問題，過一段時間才能恢復(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ā)的條件

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

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

我自定義了一個異常 NotExistResourceErr ， 觸發(fā)這個異常我才進(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):
    """
    定義一個自定義的異常
    """
    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é)果如下： 可以看出 重試了三次，第三次后 拋出了異常

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

@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)

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

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

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

retry_if_result 這里可以傳入一個函數(shù)，這個函數(shù)接收 might_return_none 返回值， 然后 這個函數(shù) 返回 True 就會 觸發(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ù)

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

retry_error_callback 這里可以傳入一個回調(diào) 函數(shù) ， 回函函數(shù)中有一個參數(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 錯誤處理

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

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

from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type, retry_if_result
class NotExistResourceErr(Exception):
    """
    定義一個自定義的異常
    """
    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 整個程序 就會掛掉，如下面的報錯：

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 推薦看下 作者的對這個庫介紹

做了一些翻譯，感興趣的同學(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ān)于Python重試庫的文章。這個庫被設(shè)計用來在一個任務(wù)發(fā)生故障時重試執(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.

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

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 就這樣誕生了，

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

nifty /?n?ft?/

niftier 比較級

niftiest 最高級

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

• Bridgeport was a pretty nifty place.

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

• It was a nifty arrangement, a perfect partnership.

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

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è)計出更簡單的東西了。顯然，這是一個相當(dāng)罕見的情況，因?yàn)槿藗兺ǔＯＭ谥卦囍g等待一段時間。為此，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:

或者可以用一個簡單的指數(shù)回退法來代替。

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é)合起來，等待在一個時間間隔內(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í)例時，例如網(wǎng)絡(luò)通信錯誤，才每秒重試執(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.

你可以通過使用|或&二進(jìn)制操作符輕松地組合幾個條件。它們被用來在引發(fā)IOError異常或沒有返回結(jié)果時使代碼重試。此外，還可以用stop關(guān)鍵字參數(shù)添加一個停止條件。它允許指定一個與要停止的異常的函數(shù)結(jié)果無關(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ù)式方法使得它可以很容易和干凈地用簡單的二進(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也可以在沒有裝飾器的情況下使用，通過使用Retrying對象，實(shí)現(xiàn)其主要行為并使用其調(diào)用方法。這允許調(diào)用任何具有不同重試條件的函數(shù)，或重試任何完全不使用裝飾器的代碼–如來自外部庫的代碼。

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ù)使用該對象，而不需要每次都創(chuàng)建一個新的對象，從而節(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!

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

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重試庫 Tenacity詳解(推薦)的文章就介紹到這了,更多相關(guān)Python Tenacity 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！