python tenacity 重试库
作者：互联网
Tenacity 重试库
tenacity 是一个很强大的重试库，前身是 retrying ，因为 retrying 停止维护了，所以有人弄出了这个库。
英文文档地址：https://tenacity.readthedocs.io/en/latest/
安装：
pip install tenacity
1. 基础使用
from tenacity import retry
@retry
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
当函数抛出异常时，
@retry会无限重试
2. 停止重试
规定次数后停止
from tenacity import stop_after_attempt
@retry(stop=stop_after_attempt(3))
def stop_after_3_attempts():
print("Stopping after 3 attempts")
raise Exception
stop_after_3_attempts()
加上限制条件，如果函数一直抛出异常，则只会重试3次
规定多久后停止
from tenacity import stop_after_attempt, stop_after_delay
@retry(stop=stop_after_delay(5))
def stop_after_5_s():
print("Stopping after 5 seconds")
raise Exception
stop_after_5_s()
会在 5 秒钟内一直重试，时间一到就会停止
多条件判断
from tenacity import stop_after_attempt, stop_after_delay
@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
stop_after_10_s_or_5_retries()
使用
|条件，可以将多个条件一起使用。如果任意一个条件满足，就会停止重试。
3. 重试的时间间隔
固定时间间隔
from tenacity import stop_after_attempt, stop_after_delay, wait_fixed
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Wait 2 second between retries")
raise Exception
wait_2_s()
每次重试之后，延迟 2 秒再进行下一次重试
随机时间间隔
from tenacity import stop_after_attempt, stop_after_delay, wait_fixed, wait_random
@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("Randomly wait 1 to 2 seconds between retries")
raise Exception
wait_random_1_to_2_s()
指数增长的时间时间
from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential
@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
wait_exponential_1()
以指数增长的时间间隔，每次重试后，间隔时间以指数增长。
计算公式：
2^x * multiplier，x是重试的次数-1。计算后的值要和min, max对比，如果值在这个范围里，则使用计算出的值。如果值大于范围，则使用max，如果值小于这个范围，则使用min
固定时间+随机时间
from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
raise Exception
wait_fixed_jitter()
给不同的次数分配不同的等待时间
from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential,wait_chain
@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
wait_fixed_chained()
4. 是否重试
4.1 根据错误类型重试
from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type
class ClientError(Exception):
"""Some type of client error."""
@retry(retry=retry_if_exception_type(IOError))
def might_io_error():
print("如果错误类型是 IOError, 则重试；其他错误类型则不会进行重试")
raise Exception
@retry(retry=retry_if_not_exception_type(ClientError))
def might_client_error():
print("如果错误类型不是 ClientError, 则进行重试；如果是 ClientError，则不会重试")
raise ClientError('')
might_client_error()
might_io_error()
4.2 根据返回值进行重试
from tenacity import retry, retry_if_result
def judge_result(x):
"""Return True if value is None"""
return x is None
@retry(retry=retry_if_result(judge_result))
def might_return_none():
print("Retry with no wait if return value is None")
return None
might_return_none()
上面的例子中，我们
might_return_none()函数的返回值会传递给judge_result()函数。如果judge_result()返回了True则会进行重试。所以这里的关键是要写一个辅助函数，判断当前函数的返回值。并将这个辅助函数传递给
retry_if_result
4.3 多条件组合
from tenacity import retry, retry_if_result, retry_if_exception_type
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")
return IOError('IO Error')
might_return_none()
否则任意一个条件满足，都会重试。
5. 错误处理
5.1 重新抛出原始异常
from tenacity import retry, retry_if_result, retry_if_exception_type, stop_after_attempt
class MyException(Exception):
pass
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Fail")
try:
raise_my_exception()
except Exception as e:
print(e)
reraise参数可以设置重新抛出原始的异常。即当最后一次重试依然有异常时，则重新抛出原始异常，而不是默认的RetryError
5.2 before 、 after
import logging
import sys
from tenacity import retry, stop_after_attempt, before_log
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
raise Exception("Fail")
raise_my_exception()
before可以在重试之前，做一些事情。这里用来记录 log。导入了一个before_log的函数。
同理，after:
import logging
import sys
from tenacity import retry, stop_after_attempt, after_log
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception():
raise Exception("Fail")
raise_my_exception()
6. 改变参数
from tenacity import retry, stop_after_attempt, retry_if_result
@retry(stop=stop_after_attempt(3))
def raise_my_exception():
print('retrying.')
raise Exception("Fail")
try:
raise_my_exception.retry_with(stop=stop_after_attempt(4))() # 运行时可以改变参数
except Exception:
pass
也可以生成一个对象，将想要重试的函数作为参数传入：
from tenacity import retry, stop_after_attempt, retry_if_result, Retrying
def never_good_enough(arg1):
print(arg1)
raise Exception('Error')
def try_never_good_enough(max_attempts=3):
retryer = Retrying(stop=stop_after_attempt(max_attempts), reraise=True) # retryer 对象
retryer(never_good_enough, 'I really do try') # 将要重试的函数和参数传递进去
try_never_good_enough()
7. 异步
可以支持 async 函数：
import asyncio
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3), reraise=True)
async def my_async_function(seconds):
await asyncio.sleep(seconds)
print(f'sleep {seconds}')
raise Exception('Async Error')
loop = asyncio.new_event_loop()
loop.run_until_complete(my_async_function(1))
标签：retry,raise,python,after,stop,重试,tenacity,wait 来源： https://www.cnblogs.com/wztshine/p/16497702.html