tenacity

重试包，简化任意任务重试行为。

0x01 安装

pip install tenacity

0x02 基本使用

from tenacity import *

@retry
def never_giveup():  # 一直重试输出，忽略异常
    print('retry forever ignoring exceptions')
    raise Exception

@retry(stop=stop_after_attempt(7)) # 设置重试次数
def stop_after_7_attempts():
    print("Stopping after 7 attempts")
    raise Exception

@retry(stop=stop_after_delay(10)) # 设置重试超时时间
def stop_after_10_s():
    print("Stopping after 10 seconds")
    raise Exception

@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))  # 超时时间和次数组合
def stop_after_10_s_or_5_retries():
    print("Stopping after 10 seconds or 5 retries")
    raise Exception

@retry(wait=wait_fixed(2))   # 重试之间间隔2秒
def wait_2_s():
    print("Wait 2 second between retries")
    raise Exception

@retry(wait=wait_exponential(multiplier=1, min=4, max=10))  # 重试
def wait_exponential_1():
    print("Wait 2^x * 1 second between each retry starting with 4 seconds, then up to 10 seconds, then 10 seconds afterwards")
    raise Exception

@retry(wait=wait_exponential(multiplier=1, min=4, max=10)) # 等待2^x * 1秒
def wait_exponential_1():
    print("Wait 2^x * 1 second between each retry starting with 4 seconds, then up to 10 seconds, then 10 seconds afterwards")
    raise Exception

@retry(wait=wait_fixed(3) + wait_random(0, 2)) # 重试间隔3到5随机
def wait_fixed_jitter():
    print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
    raise Exception

@retry(wait=wait_random_exponential(multiplier=1, max=60))  # 随机指数级间隔
def wait_exponential_jitter():
    print("Randomly wait up to 2^x * 1 seconds between each retry until the range reaches 60 seconds, then randomly up to 60 seconds afterwards")
    raise Exception

@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
                       [wait_fixed(7) for i in range(2)] +
                       [wait_fixed(9)]))  # 更复杂的间隔链
def wait_fixed_chained():
    print("Wait 3s for 3 attempts, 7s for the next 2 attempts and 9s for all attempts thereafter")
    raise Exception

0x03 是否重试

from tenacity import *

class ClientError(Exception):
    """Some type of client error."""

@retry(retry=retry_if_exception_type(IOError))
def might_io_error():
    print("Retry forever with no wait if an IOError occurs, raise any other errors")
    raise Exception

# 根据异常类型决定
@retry(retry=retry_if_not_exception_type(ClientError))
def might_client_error():
    print("Retry forever with no wait if any error other than ClientError occurs. Immediately raise ClientError.")
    raise Exception


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

# 根据结果判断是否重试
@retry(retry=retry_if_result(is_none_p))
def might_return_none():
    print("Retry with no wait if return value is None")

# 还支持以下判断
retry_if_exception
retry_if_exception_type
retry_if_not_exception_type
retry_unless_exception_type
retry_if_result
retry_if_not_result
retry_if_exception_message
retry_if_not_exception_message
retry_any
retry_all



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

# 结合多个条件
@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
    print("Retry forever ignoring Exceptions with no wait if return value is None")

# 显式重试
@retry
def do_something():
    result = something_else()
    if result == 23:
       raise TryAgain

0x04 错误处理

@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
    raise MyException("Fail")

try:
    raise_my_exception()
except MyException:
    # timed out retrying
    pass

0x05 日志

import logging
import sys

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)

logger = logging.getLogger(__name__)

# 事前 日志
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
    raise MyException("Fail")

# 事后 日志
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception2():
    raise MyException("Fail")

# 只记录将被重试的日志
@retry(stop=stop_after_attempt(3),
       before_sleep=before_sleep_log(logger, logging.DEBUG))
def raise_my_exception3():
    raise MyException("Fail")

0x06 统计

@retry(stop=stop_after_attempt(3))
def raise_my_exception():
    raise MyException("Fail")

try:
    raise_my_exception()
except Exception:
    pass

print(raise_my_exception.retry.statistics)

0x07 回调

def return_last_value(retry_state):
    """return the result of the last call attempt"""
    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=return_last_value,
       retry=retry_if_result(is_false))
def eventually_return_false():
    return False

xxxx更多