QZQ的小世界!

  • 首页
你好!
这里是QZQ的博客站!
  1. 首页
  2. 未分类
  3. 正文

Tenacity 优雅地重试

2025年4月4日 90点热度 0人点赞 0条评论

7.3 少有人知的 Python “重试机制” — Python黑魔法手册 1.0.0 documentation (iswbm.com)

为了避免由于一些网络或其他不可控因素,而引起的功能性问题。比如在发送请求时,会因为网络不稳定,往往会有请求超时的问题。

这种情况下,我们通常会在代码中加入重试的代码。重试的代码本身不难实现,但如何写得优雅、易用,是我们要考虑的问题。

这里要给大家介绍的是一个第三方库 - Tenacity ,它实现了几乎我们可以使用到的所有重试场景,比如:

  1. 在什么情况下才进行重试?

  2. 重试几次呢?

  3. 重试多久后结束?

  4. 每次重试的间隔多长呢?

  5. 重试失败后的回调?

在使用它之前 ,先要安装它

$ pip install tenacity

Intro

无条件重试,重试之间无间隔

from tenacity import retry

@retry
def test_retry():
    print("等待重试,重试无间隔执行...")
    raise Exception

test_retry()

参数

无条件重试,但是在重试之前要等待 2 秒

from tenacity import retry, wait_fixed

# 无条件重试,但是在重试之前要等待 2 秒
@retry(wait=wait_fixed(2))

# 只重试7 次
@retry(stop=stop_after_attempt(7))

# 重试 10 秒后不再重试
@retry(stop=stop_after_delay(10))

# 两个条件满足一个就结束重试
@retry(stop=(stop_after_delay(10) | stop_after_attempt(7)))

# 在出现特定错误/异常(比如请求超时)的情况下,再进行重试
@retry(retry=retry_if_exception_type(exceptions.Timeout))

自定义条件

在满足自定义条件时,再进行重试。

如下示例,当 test_retry 函数返回值为 False 时,再进行重试

from tenacity import retry, stop_after_attempt, retry_if_result

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3),
       retry=retry_if_result(is_false))
def test_retry():
    return False

test_retry()

其他事项

多个条件注意顺序

如果想对一个异常进行重试,但是最多重试3次。

下面这个代码是无效的,因为它会一直重试,重试三次的限制不会生效,因为它的条件是有顺序的,在前面的条件会先被走到,就永远走不到后面的条件。

import time
from requests import exceptions
from tenacity import retry, retry_if_exception_type, stop_after_attempt

@retry(retry=retry_if_exception_type(exceptions.Timeout), stop=stop_after_attempt(3))
def test_retry():
    time.sleep(1)
    print("retry")
    raise exceptions.Timeout

test_retry()

如果你把 stop_after_attempt 写到前边,就没有问题了。

重试后错误重新抛出

当出现异常后,tenacity 会进行重试,若重试后还是失败,默认情况下,往上抛出的异常会变成 RetryError,而不是最根本的原因。

因此可以加一个参数(reraise=True),使得当重试失败后,往外抛出的异常还是原来的那个。

@retry(stop=stop_after_attempt(7), reraise=True)

设置回调函数

当最后一次重试失败后,可以执行一个回调函数

from tenacity import *

def return_last_value(retry_state):
    print("执行回调函数")
    return retry_state.outcome.result()  # 表示返回原函数的返回值

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3),
       retry_error_callback=return_last_value,
       retry=retry_if_result(is_false))
def test_retry():
    print("等待重试中...")
    return False

print(test_retry())

[文章导入自 http://qzq-go.notion.site/2eb54a552b7d48ccbd97b032db7ca343 访问原文获取高清图片]

本作品采用 知识共享署名-非商业性使用 4.0 国际许可协议 进行许可
标签: IT技术 Python 零碎 python 小知识
最后更新:2025年4月3日

QZQ

一只涉猎广泛的技术爱好者,绝赞养猫中~

点赞
< 上一篇
下一篇 >

归档

  • 2025 年 4 月
  • 2025 年 3 月
  • 2025 年 2 月
  • 2025 年 1 月
  • 2024 年 12 月
  • 2024 年 11 月

分类

  • 技术
  • 未分类

COPYRIGHT © 2024 QZQ的小世界!. ALL RIGHTS RESERVED.

Theme Kratos Made By Seaton Jiang