Python 中的 retrying 库

Posted 2022-03-27 A-L-Kun

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python 中的 retrying 库相关的知识，希望对你有一定的参考价值。

retrying 模块
- 一、简介
- 二、使用方法
  - 1、无参数
  - 2、有参数

retrying 模块

一、简介

retrying是一个python的重试包，可以用来自动重试一些可能运行失败的程序段，retrying提供一个装饰器函数retry，被装饰的函数就会在运行失败的情况下重新执行，默认只要一直报错就会不断重试

安装：

pip install retrying

二、使用方法

1、无参数

使用 retry，默认的行为会一直重试，没有时间等待，直到不再报错为止

from retrying import retry
from random import randint


@retry  # 装饰器
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2、有参数

2.1 stop_max_attempt_number

stop_max_attempt_number：用来设定最大的尝试次数，超过该次数就会停止

from retrying import retry
from random import randint


@retry(stop_max_attempt_number=1)
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2.2 stop_max_delay

stop_max_delay：从被装饰的函数开始执行的时间点开始到函数成功运行结束或失败报错中止的时间点，单位：毫秒

from retrying import retry
from random import randint


@retry(stop_max_delay=1)
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2.3 wait_fixed

wait_fixed：设置在两次retrying之间的停留时间

from retrying import retry
from random import randint


@retry(wait_fixed=1)
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2.4 wait_random_[min/max]

wait_random_[min/max]：两次retrying之间的停留最短/最长时间

from retrying import retry
from random import randint


@retry(wait_random_min=1, wait_random_max=9)
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2.5 wait_exponential_[multiplier/max]

wait_exponential_[multiplier/max]：以指数的形式产生两次retrying之间的停留时间, 产生的值为2^previous_attempt_number * wait_exponential_multiplier, previous_attempt_number是前面已经retry的次数, 如果产生的这个值超过了wait_exponential_max的大小, 那么之后两个retrying之间的停留值都为wait_exponential_max

from retrying import retry
from random import randint


@retry(wait_exponential_multiplier=100, wait_exponential_max=10000)
def get_random():
    int_r = randint(0, 100)
    if int_r > 0:
        print(f"该随机数等于int_r")
        raise IOError("该随机数大于0")
    else:
        return int_r


print(f"该随机数等于get_random()")

2.6 其他可用参数

def __init__(self,
             stop=None, wait=None,  # 停止和等待时间
             stop_max_attempt_number=None,  # 最大尝试次数
             stop_max_delay=None,  # 最大运行时间
             wait_fixed=None,   # 延迟时间 
             wait_random_min=None, wait_random_max=None,  # 随机等待时间
             wait_incrementing_start=None,   wait_incrementing_increment=None,  # 运行下次，增加延时
             wait_exponential_multiplier=None, wait_exponential_max=None,  # 等待时间按指数形式增长
             retry_on_exception=None,  
             retry_on_result=None,
             wrap_exception=False,
             stop_func=None,  # 每次完成后运行的新函数
             wait_func=None,
             wait_jitter_max=None):