简介

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

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

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

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

推荐使用 Tenacity

安装

pip install retrying
pip install tenacity

Retrying

初试

重试无数次直至成功

import random
from retrying import retryn = 0  # 记录重试次数@retry
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise Exceptionelse: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 retryn = 0  # 记录重试次数@retry(stop_max_attempt_number=3)  # 最大重试次数
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise Exceptionelse: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 retryn = 0  # 记录重试次数@retry(stop_max_delay=3000)  # 最大重试时间为3s
def do_something_unreliable():global nn += 1raise Exceptionstart = 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 retryn = 0  # 记录重试次数
start = time.time()  # 开始时间@retry(wait_fixed=500)  # 重试间隔0.5秒
def do_something_unreliable():global n, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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 retryn = 0  # 记录重试次数
start = time.time()  # 开始时间@retry(wait_random_min=500, wait_random_max=1000)  # 重试间隔0.5-1秒
def do_something_unreliable():global n, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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 retryn = 0  # 记录重试次数
start = time.time()  # 开始时间@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)  # 指数级重试间隔=2^x*1000ms，10s封顶
def do_something_unreliable():global n, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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 retryn = 0  # 记录重试次数@retry(retry_on_exception=lambda x: isinstance(x, IOError))  # 自定义异常重试
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise IOErrorelse: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 retryn = 0  # 记录重试次数@retry(retry_on_exception=lambda x: isinstance(x, IOError), wrap_exception=True)  # 自定义异常结果
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise IOErrorelse:print(f'tried {n} times')n = 0  # 重置重试次数if random.randint(0, 2) > 1:raise IndexErrorfor 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 retryn = 0  # 记录重试次数@retry(retry_on_result=lambda x: x is None)  # 自定义异常重试
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1return Noneelse:return nfor 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 retryn = 0  # 记录重试次数@retry
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise Exceptionelse: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_attemptn = 0  # 记录重试次数@retry(stop=stop_after_attempt(3))  # 最大重试次数
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise Exceptionelse: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_delayn = 0  # 记录重试次数@retry(stop=stop_after_delay(3))  # 最大重试时间为3s
def do_something_unreliable():global nn += 1raise Exceptionstart = 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_delayn = 0  # 记录重试次数@retry(stop=stop_after_attempt(300000) | stop_after_delay(3))  # 最大重试30w次 或 最长重试3秒
def do_something_unreliable():global nn += 1raise Exceptionstart = 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_fixedn = 0  # 记录重试次数
start = time.time()  # 开始时间@retry(wait=wait_fixed(0.5))  # 重试间隔0.5秒
def do_something_unreliable():global n, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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_randomn = 0  # 记录重试次数
start = time.time()  # 开始时间@retry(wait=wait_random(min=0.5, max=1))  # 重试间隔0.5-1秒
def do_something_unreliable():global n, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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_randomn = 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, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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_exponentialn = 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, startif random.randint(0, 3) > 1:n += 1raise Exceptionelse: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_chainn = 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 nif n == 10:returnn += 1end = time.time()print(f'tried {n} times, pass {end - start} seconds')raise Exceptiondo_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_typen = 0  # 记录重试次数@retry(retry=retry_if_exception_type(IOError))  # 自定义异常重试
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1raise IOErrorelse: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_resultn = 0  # 记录重试次数@retry(retry=retry_if_result(lambda x: x is None))  # 自定义异常结果
def do_something_unreliable():global nif random.randint(0, 10) > 1:n += 1return Noneelse:return nfor 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 += 1if n == 10:raise TryAgainreturn nn = 0
while n <= 15:try:n = do_something_unreliable(n)except:n += 1else: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_delayn = 0
start = time.time()@retry(reraise=True, stop=stop_after_attempt(3))
def do_something_unreliable():"""超过最大重试次数后抛出异常"""global nn += 1raise 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_loglogging.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_loglogging.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_loglogging.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_resultn = 0def 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 nn += 1return nprint(do_something_unreliable())
# 3

上下文管理器

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

from tenacity import Retrying, RetryError, stop_after_attempttry:for attempt in Retrying(stop=stop_after_attempt(3)):with attempt:raise Exception('Fail')
except RetryError:pass

协程

from tenacity import AsyncRetrying, RetryError, stop_after_attemptasync 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重试库Retrying和Tenacity相关推荐

1. python之Tenacity重试库的使用

   目录 安装 Tenacity 导入 Tenacity 模块 为需要重试的函数添加装饰器 调用重试函数 常用的重试方法 Tenacity 示例 Tenacity 是一个用于 Python 的重试库,它提 ...

2. Python 中最强大的错误重试库

   作者 | 费弗里 来源丨Python大数据分析 1 简介 我们在编写程序尤其是与网络请求相关的程序,如调用web接口.运行网络爬虫等任务时,经常会遇到一些偶然发生的请求失败的状况,这种时候如果我们仅仅 ...

3. python 重试—retry库的使用和自定义超时retry

   python 重试-retry库的使用和自定义超时retry 引言 retry 库 安装 使用 参数说明 自定义retry 引言 当我们写代码自动化测试代码时,由于页面读取.数据刷新等造成的页面元素不 ...

4. python clicknium 库自动化千牛桌面端

   python clicknium 库自动化千牛桌面端 千牛是阿里巴巴集团卖家工作台,商家经营的必备工具,今天我们使用python来自动化千牛桌面端. clicknium 是基于 python 实现的一 ...

5. python安装库备忘

   python安装库备忘 参考 pip注意事项 python库备忘 参考 Python开发之pip使用详解 pypi pypi pip注意事项 默认安装库时按最新版本安装,可能把以前的库冲掉,造成版本不 ...

6. 安装和测试Python第三方库20200628

   安装和测试Python第三方库 #!/usr/bin/env python # coding:utf-8import sysprint("Python解释器在磁盘上的存储路径:", ...

7. QGIS中安装Python第三方库

   在QGIS自带的python中安装第三方库 如果安装了QGIS,在所有程序中搜OSGeo4W Shell,以管理员身份打开,可以安装第三方库 环境设置 错误尝试: 直接安装,不是QGIS中的pytho ...

8. Python requests库核心源码解析

   Requests is an elegant and simple HTTP library for Python, built for human beings. Python requests是最 ...

9. Python 标准库之 xml.etree.ElementTree xml解析

   Python 标准库之 xml.etree.ElementTree Python中有多种xml处理API,常用的有xml.dom.*模块.xml.sax.*模块.xml.parser.expat模块和 ...

最新文章

1. mysql性能优化1
2. leetcode算法题--0～n-1中缺失的数字
3. LSMW批处理使用方法(08)_步骤8、9
4. php连接数据库封装函数,PHP基于MySQLI函数封装的数据库连接工具类【定义与用法】...
5. java： jmap 查看内存信息
6. 约瑟夫问题（丢手帕问题）的java实现
7. 第二层$.get()、$.post() 方法使用（三）
8. C语言解力扣461.汉明距离
9. 读书笔记——并行处理器架构
10. 智和网管：深入国产化需求，扎根网络运维安全
11. BUUCTF WEB PIAPIAPIA1
12. murmur3 php,murmur: 更快更好的哈希函数
13. Python爬取的微信好友信息里我看到了自律 | CSDN博文精选
14. matlab relieff函数,数据挖掘 ReliefF和K-means算法的应用
15. 使用react进行项目开发
16. 基于Android平台的手机安全助手的设计与实现
17. 接触角测量的常用测量法
18. 张飞老师硬件第五部视频整理——模拟电路
19. 服务器如何合理设置虚拟内存,服务器虚拟内存设置多大合适
20. Linux 文件目录压缩与解压命令

热门文章

1. 直销银行二三事之敏捷开户
2. opencv实战，钢板焊接点寻找1
3. 取消iostream与stdio之间的绑定
4. 人脸伪造DFDC数据集下载 附网盘地址
5. Microsoft Word的学习
6. 一、STM32程序下载软件_FlyMCU
7. Csvmapper操作csv文件
8. 新人制作机器人的7大误区
9. 2017微信公开课·张小龙演讲全文(上)
10. 5个高质量图片素材网站