[Python知识库]Python：retrying与tenacity模块失败重跑库

简介：实际编程中网络请求或者IO等操作中存在偶然发生失败的状况，需要在程序逻辑添加错误重试的策略。retrying、tenacity模块均为失败重跑库，各有优劣。

安装：

pip install retrying
pip install tenacity

相关攻略：

python：超时跳过

retrying参数详解：

retry                                            # 不加参数表示无限重试
retry(wait_fixed = 100)                          # 设置重试间隔时长，单位:ms，1000ms = 1s
retry(wait_random_min=100,wait_random_max=200)   # 随机重试间隔，将在100ms - 200ms内
retry(stop_max_attempt_number = 4)               # 最大重试次数，超过后正常抛出异常
retry(stop_max_delay=1)                          # 最大延迟时长，1ms内未满足条件则抛出异常
retry(retry_on_exception = 自定义函数)           # 当发生指定异常时会执行函数
retry(retry_on_result=自定义函数)                # 每次都会执行函数，当返回返回True就重试，否则异常退出

retryingDemo源码：

# -*- coding: utf-8 -*-
# time: 2022/5/8 0:47
# file: retrying_demo.py
# 公众号: 玩转测试开发
import random
from retrying import retry


class Retry:
    """
    参数说明：
    retry                                            # 不加参数表示无限重试
    retry(wait_fixed = 100)                          # 设置重试间隔时长，单位:ms，1000ms = 1s
    retry(wait_random_min=100,wait_random_max=200)   # 随机重试间隔，将在100ms - 200ms内
    retry(stop_max_attempt_number = 4)               # 最大重试次数，超过后正常抛出异常
    retry(stop_max_delay=1)                          # 最大延迟时长，1ms内未满足条件则抛出异常
    retry(retry_on_exception = 自定义函数)           # 当发生指定异常时会执行函数
    retry(retry_on_result=自定义函数)                # 每次都会执行函数，当返回返回True就重试，否则异常退出
    """

    @retry
    def run1(self):
        """
        不加参数表示无限重试
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(wait_fixed=100)
    def run2(self):
        """
        wait_fixed: 设置重试间隔时长，单位:ms，1000ms = 1s
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(wait_random_min=100, wait_random_max=200)
    def run3(self):
        """
        随机重试间隔，将在100ms - 200ms内
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(stop_max_attempt_number=4)
    def run4(self):
        """
        最大重试次数，超过后正常抛出异常
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(stop_max_delay=1)
    def run5(self):
        """
        最大延迟时长，1ms内未满足条件则抛出异常
        :return:
        """
        a = random.randint(0, 1000)
        print(a)
        if a >= 5:
            raise Exception

    def f1(r):
        return isinstance(r, ZeroDivisionError)

    @retry(retry_on_exception=f1)
    def run6(self):
        """
        retry(retry_on_exception = 自定义函数): 当发生指定异常时会执行函数
        :return:
        """
        r = random.randint(0, 1)
        print(r)
        if r == 0:  # 不遇到ZeroDivisionError则会退出程序，遇到ZeroDivisionError则重试。
            raise 1 / r

    def f2(r):
        if r > 50:
            return False
        else:
            return True

    @retry(retry_on_result=f2)
    def run7(self):
        """
        retry(retry_on_result = 自定义函数): 每次都会执行函数，当返回返回True就重试，否则异常退出
        :return:
        """
        r = random.randint(0, 100)
        print(r)
        return r


if __name__ == '__main__':
    retrying_retry = Retry()
    # retrying_retry.run1()
    # retrying_retry.run2()
    # retrying_retry.run3()
    # retrying_retry.run4()
    # retrying_retry.run5()
    # retrying_retry.run6()
    retrying_retry.run7()

tenacity参数详解：

retry                                                        # 不加参数表示无限重试
retry(stop=stop_after_attempt(3))                            # 最大重试次数，超过后正常抛出异常
retry(stop=stop_after_delay(1)                               # 设置重试最大超时时长
retry(stop=(stop_after_delay(5) | stop_after_attempt(10)))   # 组合重试停止条件：最多运行5秒 或者最多重试10次。
retry(wait=wait_fixed(1))                                    # 设置重试间隔时长，注意单位为:秒。
retry(wait=wait_random(min=0.1, max=0.5))                    # 设置重试随机间隔时长，注意单位为:秒。
retry(retry=retry_if_exception_type(ZeroDivisionError))      # 自定义是否触发重试：含有指定的错误类型，如ZeroDivisionError
retry(retry=retry_if_not_exception_type(ZeroDivisionError))  # 自定义是否触发重试：未含有指定的错误类型
retry=retry_if_result(f1)：# 每次都会执行函数，当返回返回True就重试，否则异常退出
retry.statistics：# 属性查看其历经的错误重试统计记录结果

tenacityDemo源码：

# -*- coding: utf-8 -*-
# time: 2022/4/30 23:40
# file: test.py
# 公众号: 玩转测试开发
import random
import time
from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, \
    retry_if_exception_type, retry_if_not_exception_type, retry_if_result


class Retry:
    """
    参数说明：
    retry                                                        # 不加参数表示无限重试
    retry(stop=stop_after_attempt(3))                            # 最大重试次数，超过后正常抛出异常
    retry(stop=stop_after_delay(1)                               # 设置重试最大超时时长
    retry(stop=(stop_after_delay(5) | stop_after_attempt(10)))   # 组合重试停止条件：最多运行5秒 或者最多重试10次。
    retry(wait=wait_fixed(1))                                    # 设置重试间隔时长，注意单位为:秒。
    retry(wait=wait_random(min=0.1, max=0.5))                    # 设置重试随机间隔时长，注意单位为:秒。
    retry(retry=retry_if_exception_type(ZeroDivisionError))      # 自定义是否触发重试：含有指定的错误类型，如ZeroDivisionError
    retry(retry=retry_if_not_exception_type(ZeroDivisionError))  # 自定义是否触发重试：未含有指定的错误类型
    retry=retry_if_result(f1)：# 每次都会执行函数，当返回返回True就重试，否则异常退出
    retry.statistics：# 属性查看其历经的错误重试统计记录结果
    """

    @retry
    def run1(self):
        """
        不加参数表示无限重试
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(stop=stop_after_attempt(3))
    def run2(self):
        """
        设置最大重试次数
        :return:
        """
        a = random.randint(0, 100)
        print(a)
        if a >= 5:
            raise Exception

    @retry(stop=stop_after_delay(1))
    def run3(self):
        """
        设置重试最大超时时长
        :return:
        """
        during = random.random()
        time.sleep(during)
        print(f"during:{during}, 已运行时间：{time.time() - start} ")
        raise Exception

    @retry(stop=(stop_after_delay(5) | stop_after_attempt(10)))
    def run4(self):
        """
        组合重试停止条件：最多运行5秒 或者最多重试10次。
        :return:
        """
        during = random.random()
        time.sleep(during)
        print(f"during:{during}, 已运行时间：{time.time() - start} ")
        raise Exception

    @retry(wait=wait_fixed(1))
    def run5(self):
        """
        设置重试间隔时长，注意单位为:秒。
        :return:
        """
        result = random.randint(0, 100)
        print(result)
        if result > 10:
            raise Exception

    @retry(wait=wait_random(min=0.1, max=0.5))
    def run6(self):
        """
        设置重试随机间隔时长，注意单位为:秒。
        :return:
        """
        result = random.randint(0, 100)

        print(f"当前数值：{result}，已运行时间：{time.time() - start} ")
        if result > 5:
            raise Exception

    @retry(retry=retry_if_exception_type(ZeroDivisionError))
    def run7(self):
        """
        自定义是否触发重试：含有指定的错误类型，如ZeroDivisionError
        :return:
        """
        r = random.randint(0, 1)
        print(r)
        if r == 0:  # 不遇到ZeroDivisionError则会退出程序，遇到ZeroDivisionError则重试。
            raise 1 / r

    @retry(retry=retry_if_not_exception_type(ZeroDivisionError))
    def run8(self):
        """
        自定义是否触发重试：未含有指定的错误类型，如ZeroDivisionError。抛出其他错误则继续运行
        :return:
        """
        r = random.randint(1, 100)
        print(r)
        if r < 10:
            return ZeroDivisionError
        else:
            raise Exception

    def f1(r):
        if r > 50:
            return False
        else:
            return True

    @retry(retry=retry_if_result(f1))
    def run9(self):
        """
        retry=retry_if_result(f1)：每次都会执行函数，当返回返回True就重试，否则异常退出
        :return:
        """
        r = random.randint(0, 100)
        print(r)
        return r

    @retry
    def run10(self):
        """
        retry.statistics：属性查看其历经的错误重试统计记录结果
        :return:
        """
        self.run9()
        # 如 # {'start_time': 586421.812, 'attempt_number': 2, 'idle_for': 0, 'delay_since_first_attempt': 0.0}
        print(self.run9.retry.statistics)


if __name__ == '__main__':
    start = time.time()
    tenacity_retry = Retry()
    # tenacity_retry.run1()
    # tenacity_retry.run2()
    # tenacity_retry.run3()
    # tenacity_retry.run4()
    # tenacity_retry.run5()
    # tenacity_retry.run6()
    # tenacity_retry.run7()
    # tenacity_retry.run8()
    # tenacity_retry.run9()
    tenacity_retry.run10()

结论：一般的场景下retrying即可满足，写法更加简洁明了。特殊场景下使用tenacity。

微信公众号：玩转测试开发
欢迎关注，共同进步，谢谢！

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖