Python重试库Retrying和Tenacity的实现_Python

简介

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	回调函数

到此这篇关于python重试库retrying和tenacity的实现的文章就介绍到这了,更多相关python重试库retrying和tenacity内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

Python重试库Retrying和Tenacity的实现

2025年09月28日 • Python •我要评论

简介

安装

retrying

初试

最大重试次数

最大重试时间

重试间隔

随机间隔

指数级重试间隔

自定义异常重试

自定义返回重试

参数

tenacity

初试

最大重试次数

最大重试时间

组合停止条件

重试间隔

随机间隔

指数级重试间隔

重试间隔链

自定义异常重试

自定义返回重试

显式重试

重新捕获异常

重试前记日志

重试后记日志

重试失败后记日志

重试统计

自定义回调函数

上下文管理器

支持异步和协程

参数

相关文章:

Python+PyQt构建自动化定时任务执行工具详细代码示例

发表评论


验证码：