在写代码时经常会遇到对抛出异常的代码进行重试,常见于网页爬虫的代码中,使用计数器 + 循环的方式对抛出异常的代码进行捕获和重试。tenacity是使用Python装饰器模式对方法异常进行捕获,通过灵活的参数实现简单优雅的异常重试。

特性:

  1. 简单灵活的装饰模式api
  2. 可以指定重试停止条件(比如:设置重试次数)
  3. 也可以指定等待条件(比如:使用指数避让间隔重试)
  4. 自定义触发重试的Exception
  5. 自定义重试预期的返回结果
  6. 基于协程的重试

安装方式:

pip install tenacity

API使用介绍

1. @retry

给需要重试的方法加上@retry修饰器之后,方法抛出异常就会被装饰器捕获到并进行重试,异常抛出时会不断重试直到方法成功返回

1
2
3
4
@retry
def never_give_up_never_surrender():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception

2. 带终止条件的retry

我们也可以给retry加一个参数设置重试n次后不再重试并抛出异常

1
2
3
4
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Stopping after 7 attempts")
raise Exception

使用@stop_after_delay 可以指定重试间隔,比如如下的例子指定10秒后重试

1
2
3
4
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("Stopping after 10 seconds")
raise Exception

可以使用 “|” 把多个条件组合起来

1
2
3
4
@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

3. 在重试前等待

使用@wait_fixed 在重试前等待固定时间

1
2
3
4
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Wait 2 second between retries")
raise Exception

随机等待1-2秒钟,这在爬虫爬网页时比较有用

1
2
3
4
@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

增加指数避让等待间

1
2
3
4
5
6
7
8
9
@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_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

看一个更复杂点的例子

1
2
3
4
5
6
@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

4. 带触发条件的retry语句

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@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

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(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")

5. 异常处理


虽然tenacity会帮我们处理异常,我们依然可以在重试失败后使用reraise来决定我们时候进行最后的尝试,使用reraise会把异常抛出交给我们的try except来处理

1
2
3
4
5
6
7
8
9
@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

6. 在retry前后增加log

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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_exception():
raise MyException("Fail")

@retry(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Fail")

7. 统计异常情况

1
2
3
4
5
6
7
8
9
10
@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)

输出如下内容:

1
{'start_time': 283085.571804807, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0002240639878436923}

8. 自定义异常回调函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from tenacity import stop_after_attempt, retry_if_result, retry

def return_last_value(retry_state):
"""return the result of the last call attempt"""
return retry_state.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

print(eventually_return_false())

输出结果为 False

项目Git地址

https://github.com/jd/tenacity