Python异常处理伴侣 -- tenacity模块使用介绍

2018年12月19日

我的微信公众号：pyquant

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

特性：

简单灵活的装饰模式api
可以指定重试停止条件（比如：设置重试次数）
也可以指定等待条件（比如：使用指数避让间隔重试）
自定义触发重试的Exception
自定义重试预期的返回结果
基于协程的重试

安装方式:

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