登录
当前位置： 首页 > 工具软件 > Tenacity > 使用案例 >

Python重试库Retrying和Tenacity

令狐和裕
2023-12-01

简介

Retrying 是一个通用重试库，用于简化任何需要重试的任务，已不再维护，功能：

  • 通用装饰器
  • 指定停止条件，如重试次数
  • 指定等待条件，如重试间的指数回退休眠
  • 自定义异常重试
  • 自定义重试预期返回的结果

Tenacity 是上述库的分支，修复了一些BUG，增加了新功能：

  • 异步协程重试
  • 上下文管理器重试
  • 组合停止条件



推荐使用 Tenacity




安装

pip install retrying
pip install tenacity




Retrying

初试

重试无数次直至成功

import random
from retrying import retry

n = 0  # 记录重试次数


@retry
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 1 times
    # tried 2 times
    # tried 4 times
    # tried 5 times
    # tried 13 times




最大重试次数

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(stop_max_attempt_number=3)  # 最大重试次数
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times and success.')
        n = 0  # 重置重试次数


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(f'tried {n} times but failed.')
        n = 0  # 重置重试次数
    # tried 3 times but failed.
    # tried 1 times and success.
    # tried 0 times and success.
    # tried 2 times and success.
    # tried 3 times but failed.




最大重试时间

单位为毫秒

import time
from retrying import retry

n = 0  # 记录重试次数


@retry(stop_max_delay=3000)  # 最大重试时间为3s
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 796060 times, pass 2.999951124191284 seconds




重试间隔

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_fixed=500)  # 重试间隔0.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 1.538625955581665 seconds
    # tried 1 times, pass 0.5115864276885986 seconds
    # tried 1 times, pass 0.5024125576019287 seconds
    # tried 0 times, pass 0.0 seconds




随机间隔

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_random_min=500, wait_random_max=1000)  # 重试间隔0.5-1秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 1 times, pass 0.7865383625030518 seconds
    # tried 1 times, pass 0.5917379856109619 seconds
    # tried 6 times, pass 4.129276990890503 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 2.2903735637664795 seconds




指数级重试间隔

x为重试次数，如第一次重试间隔2s，第二次4s，第三次8s，第四次16s（不设封顶的话）

import time
import random
from retrying import retry

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)  # 指数级重试间隔=2^x*1000ms，10s封顶
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 1 times, pass 2.0003767013549805 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 4 times, pass 24.02167558670044 seconds




自定义异常重试

只重试 IOError

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_exception=lambda x: isinstance(x, IOError))  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 5 times
    # tried 2 times
    # tried 3 times
    # tried 6 times
    # tried 2 times

只重试 IOError，其余抛出

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_exception=lambda x: isinstance(x, IOError), wrap_exception=True)  # 自定义异常结果
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数
    if random.randint(0, 2) > 1:
        raise IndexError


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(e)
    # tried 4 times
    # tried 5 times
    # tried 2 times
    # tried 6 times
    # tried 1 times
    # RetryError[Attempts: 2, Error:
    #   File "C:\Users\Administrator\Envs\test\lib\site-packages\retrying.py", line 200, in call
    #     attempt = Attempt(fn(*args, **kwargs), attempt_number, False)
    #   File "D:/mycode/xxx.py", line 18, in do_something_unreliable
    #     raise IndexError
    # ]




自定义返回重试

返回为 None 则重试

import random
from retrying import retry

n = 0  # 记录重试次数


@retry(retry_on_result=lambda x: x is None)  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        return None
    else:
        return n


for i in range(5):
    print(f'tried {do_something_unreliable()} times')
    n = 0  # 重置重试次数
    # tried 10 times
    # tried 8 times
    # tried 0 times
    # tried 10 times
    # tried 3 times




参数

参数功能
无重试直至成功
stop_max_attempt_number最大重试次数
stop_max_delay最大重试时间（毫秒）
wait_fixed重试间隔（毫秒）
wait_random_min 和 wait_random_max随机重试间隔（毫秒）
wait_exponential_multiplier 和 wait_exponential_max指数级重试间隔（毫秒）
retry_on_exception自定义异常重试
wrap_exception是否抛出其余重试
retry_on_result自定义异常结果




Tenacity

初试

import random
from tenacity import retry

n = 0  # 记录重试次数


@retry
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 1 times
    # tried 0 times
    # tried 4 times
    # tried 7 times
    # tried 3 times




最大重试次数

import random
from tenacity import retry, stop_after_attempt

n = 0  # 记录重试次数


@retry(stop=stop_after_attempt(3))  # 最大重试次数
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise Exception
    else:
        print(f'tried {n} times and success.')
        n = 0  # 重置重试次数


for i in range(5):
    try:
        do_something_unreliable()
    except Exception as e:
        print(f'tried {n} times but failed.')
        n = 0  # 重置重试次数
    # tried 1 times and success.
    # tried 3 times but failed.
    # tried 3 times but failed.
    # tried 2 times and success.
    # tried 2 times and success.




最大重试时间

单位为秒

import time
from tenacity import retry, stop_after_delay

n = 0  # 记录重试次数


@retry(stop=stop_after_delay(3))  # 最大重试时间为3s
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds




组合停止条件

通过操作符 | 组合多个停止条件

import time
from tenacity import retry, stop_after_attempt, stop_after_delay

n = 0  # 记录重试次数


@retry(stop=stop_after_attempt(300000) | stop_after_delay(3))  # 最大重试30w次 或 最长重试3秒
def do_something_unreliable():
    global n
    n += 1
    raise Exception


start = time.time()
try:
    do_something_unreliable()
except:
    pass
end = time.time()
print(f'tried {n} times, pass {end - start} seconds')
# tried 206456 times, pass 2.9916257858276367 seconds




重试间隔

import time
import random
from tenacity import retry, wait_fixed

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_fixed(0.5))  # 重试间隔0.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 1 times, pass 0.5027601718902588 seconds
    # tried 2 times, pass 1.0299296379089355 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 1.5316619873046875 seconds
    # tried 4 times, pass 2.0474536418914795 seconds




随机间隔

import time
import random
from tenacity import retry, wait_random

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_random(min=0.5, max=1))  # 重试间隔0.5-1秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 3 times, pass 2.287365674972534 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 2 times, pass 1.4969894886016846 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 6 times, pass 4.51520299911499 seconds

同样可以组合（不过我觉得这样闲着蛋疼）

import time
import random
from tenacity import retry, wait_fixed, wait_random

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_fixed(0.5) + wait_random(min=0.5, max=1))  # 重试间隔1-1.5秒
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 2 times, pass 2.9294729232788086 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 0 times, pass 0.0 seconds
    # tried 3 times, pass 3.8608667850494385 seconds
    # tried 1 times, pass 1.4092319011688232 seconds




指数级重试间隔

x为重试次数，最小1s，最大10s

import time
import random
from tenacity import retry, wait_exponential

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_exponential(multiplier=1, min=1, max=10))  # 指数级重试间隔=2^x*1s，最小1s，最大10s
def do_something_unreliable():
    global n, start
    if random.randint(0, 3) > 1:
        n += 1
        raise Exception
    else:
        end = time.time()
        print(f'tried {n} times, pass {end - start} seconds')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    start = time.time()  # 更新开始时间
    # tried 0 times, pass 0.0 seconds
    # tried 6 times, pass 35.06491994857788 seconds
    # tried 2 times, pass 3.013124942779541 seconds
    # tried 1 times, pass 1.010573387145996 seconds
    # tried 0 times, pass 0.0 seconds




重试间隔链

import time
import random
from tenacity import retry, wait_fixed, wait_chain

n = 0  # 记录重试次数
start = time.time()  # 开始时间


@retry(wait=wait_chain(*[wait_fixed(0.5)] * 3 + [wait_fixed(1)] * 2 + [wait_fixed(2)]))  # 重试间隔链，先3个0.5秒，再2个1秒，之后都是2秒
def do_something_unreliable():
    global n
    if n == 10:
        return
    n += 1
    end = time.time()
    print(f'tried {n} times, pass {end - start} seconds')
    raise Exception


do_something_unreliable()
# tried 1 times, pass 0.0 seconds
# tried 2 times, pass 0.5056586265563965 seconds
# tried 3 times, pass 1.0193772315979004 seconds
# tried 4 times, pass 1.5333683490753174 seconds
# tried 5 times, pass 2.5386297702789307 seconds
# tried 6 times, pass 3.5489938259124756 seconds
# tried 7 times, pass 5.551833629608154 seconds
# tried 8 times, pass 7.559761047363281 seconds
# tried 9 times, pass 9.561469554901123 seconds
# tried 10 times, pass 11.570155143737793 seconds




自定义异常重试

只重试 IOError

import random
from tenacity import retry, retry_if_exception_type

n = 0  # 记录重试次数


@retry(retry=retry_if_exception_type(IOError))  # 自定义异常重试
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        raise IOError
    else:
        print(f'tried {n} times')
        n = 0  # 重置重试次数


for i in range(5):
    do_something_unreliable()
    # tried 5 times
    # tried 2 times
    # tried 3 times
    # tried 6 times
    # tried 2 times




自定义返回重试

返回为 None 则重试

import random
from tenacity import retry, retry_if_result

n = 0  # 记录重试次数


@retry(retry=retry_if_result(lambda x: x is None))  # 自定义异常结果
def do_something_unreliable():
    global n
    if random.randint(0, 10) > 1:
        n += 1
        return None
    else:
        return n


for i in range(5):
    print(f'tried {do_something_unreliable()} times')
    n = 0  # 重置重试次数
    # tried 10 times
    # tried 8 times
    # tried 0 times
    # tried 10 times
    # tried 3 times




显式重试

from tenacity import retry, stop_after_delay, TryAgain


@retry(stop=stop_after_delay(3))
def do_something_unreliable(n):
    """显式重试"""
    n += 1
    if n == 10:
        raise TryAgain
    return n


n = 0
while n <= 15:
    try:
        n = do_something_unreliable(n)
    except:
        n += 1
    else:
        print(n, end=' ')
# 1 2 3 4 5 6 7 8 9 11 12 13 14 15 16




重新捕获异常

超过最大重试次数或时间后抛出异常

import time
from tenacity import retry, stop_after_attempt, stop_after_delay

n = 0
start = time.time()


@retry(reraise=True, stop=stop_after_attempt(3))
def do_something_unreliable():
    """超过最大重试次数后抛出异常"""
    global n
    n += 1
    raise IOError(f'tried {n} times but failed.')


@retry(reraise=True, stop=stop_after_delay(3))
def do_something_unreliable1():
    """超过最大重试时间后抛出异常"""
    end = time.time()
    raise IOError(f'tried {end - start:.4f} seconds but failed.')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
try:
    do_something_unreliable1()
except Exception as e:
    print(e)
# tried 3 times but failed.
# tried 2.9864 seconds but failed.




重试前记日志

import sys
import logging
from tenacity import retry, stop_after_attempt, before_log

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 do_something_unreliable():
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# RetryError[<Future at 0x1b0342d7a58 state=finished raised OSError>]
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 1st time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 2nd time calling it.
# DEBUG:__main__:Starting call to '__main__.do_something_unreliable', this is the 3rd time calling it.




重试后记日志

import sys
import time
import logging
from tenacity import retry, stop_after_attempt, after_log

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

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def do_something_unreliable():
    time.sleep(0.5)
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 0.500(s), this was the 1st time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.000(s), this was the 2nd time calling it.
# DEBUG:__main__:Finished call to '__main__.do_something_unreliable' after 1.500(s), this was the 3rd time calling it.
# RetryError[<Future at 0x22b45e07a58 state=finished raised OSError>]




重试失败后记日志

import sys
import time
import logging
from tenacity import retry, stop_after_attempt, before_sleep_log

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

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), before_sleep=before_sleep_log(logger, logging.DEBUG))
def do_something_unreliable():
    time.sleep(0.5)
    raise IOError('Fail')


try:
    do_something_unreliable()
except Exception as e:
    print(e)
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# DEBUG:__main__:Retrying __main__.do_something_unreliable in 0.0 seconds as it raised OSError: Fail.
# RetryError[<Future at 0x1c840b96ac8 state=finished raised OSError>]




重试统计

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def do_something_unreliable():
    raise IOError('Fail')


try:
    do_something_unreliable()
except:
    pass
print(do_something_unreliable.retry.statistics)
# {'start_time': 756545.281, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0}

do_something_unreliable.retry.statistics["attempt_number"] 可直接获取重试次数，不用手动计算




自定义回调函数

retry_state 是 RetryCallState 的实例

from tenacity import retry, stop_after_attempt, retry_if_result

n = 0


def return_value(retry_state):
    """自定义回调函数"""
    return retry_state.outcome.result()  # 返回函数产生的最后结果或异常


@retry(stop=stop_after_attempt(3),
       retry_error_callback=return_value,
       retry=retry_if_result(lambda x: isinstance(x, int)))
def do_something_unreliable():
    global n
    n += 1
    return n


print(do_something_unreliable())
# 3




上下文管理器

结合 for 循环和上下文管理器

from tenacity import Retrying, RetryError, stop_after_attempt

try:
    for attempt in Retrying(stop=stop_after_attempt(3)):
        with attempt:
            raise Exception('Fail')
except RetryError:
    pass

协程

from tenacity import AsyncRetrying, RetryError, stop_after_attempt


async def function():
    try:
        async for attempt in AsyncRetrying(stop=stop_after_attempt(3)):
            with attempt:
                raise Exception('Fail')
    except RetryError:
        pass




支持异步和协程

import trio
import asks
import tornado
from tenacity import retry


@retry
async def my_async_function(loop):
    await loop.getaddrinfo('8.8.8.8', 53)


@retry
@tornado.gen.coroutine
def my_async_function(http_client, url):
    yield http_client.fetch(url)


@retry(sleep=trio.sleep)
async def my_async_function(loop):
    await asks.get('https://example.org')




参数

参数功能取值
无重试直至成功
stop停止条件最大重试次数 stop_after_attempt
最大重试时间 stop_after_delay
wait等待条件重试间隔 wait_fixed
随机重试间隔wait_random
指数级重试间隔 wait_exponential
重试间隔链 wait_chain
retry重试条件自定义异常 retry_if_exception_type
自定义返回 retry_if_result
reraise是否抛出异常bool
before重试前动作重试前记日志 before_log
after重试后动作重试后记日志 after_log
before_sleep重试失败后动作before_sleep_log 重试失败后记日志
retry_error_callback回调函数




参考文献

  1. Retrying GitHub
  2. Tenacity GitHub
  3. Tenacity Documentation
  4. Tornado Documentation
  5. Trio Documentation
  6. asks Documentation
  7. 少有人知的 Python “重试机制”
  8. Python利器：retrying失败、异常重试模块库
 类似资料：

相关阅读

python请求模块和连接重用Python类继承和__dict__查找[重复]通过共享库连接Python和Torch7（Lua）如何在Eclypse和pydev中添加python“库”利用Python中的mock库对Python代码进行模拟测试

相关文章

Python Pandas去重Python BS4解析库Python面试题之Python面试题汇总重汽面试【华为OD机试2023】字符串重新排序(Python)

相关问答

python中的Postgres数据库连接[重复]使用JSch库的上载重试机制最小化重复的存储库测试Python和Selenium-重用whatsapp Web会话Seleium和Geckodriver不工作python[重复]

相关文档

Python 面试题关于 Python 的面试题Keras 基于 Python 的深度学习库Python 自动化测试之实操Python 和 Pandas 数据分析教程
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.