一个python实现重试机制的简要实践

最近在写接口测试脚本时，遇到如下一个测试场景

1、A系统会创建一条数据，创建成功后会把数据推到B系统；

2、由于是两个系统之间通信，数据不会立刻从A系统同步到B系统，中间有一个短暂的时间差；

我要调试的接口有2个，一是在A系统调用一个接口，生成数据；二是在B系统调用另一个接口处理数据。

实际操作后，发现一个问题：由于调用完A接口后，会立刻调用B接口，从代码层面看，这个时间差很短，虽然A系统已经把数据生成了，但在这么短时间内还没推送到B系统，导致调用B接口时，查不到这条数据，就会报错

第一个解决方案

开始想到的解决方案是使用time.sleep()，当调用A接口后，等待一段时间，如 time.sleep(5)，死等5s，然后再调用B接口

因为等待5s后，数据一般能够从A系统推送到B系统

当然如果5s后还没有同步到B系统，调用B接口时仍然会报错，所以这并不是一个很好的解决方案

第二个解决方案

互联网冲浪一番后发现了python有一个库可以实现重试机制：tenacity

下面是找到的一些参考博客，可以看一下基础用法：https://www.cnblogs.com/wuzhibinsuib/p/13443622.html

接下来说一下自己的实验结果以及理解

它的简单用法是给需要重试的代码加上@retry修饰器，代码抛出异常会被装饰器捕获并进行重试

这里的关键是捕获到到代码抛出的异常

例1【如果报错会一直重试】

@retry
def test_retry1():
    print("等待重试.....")
    raise Exception  # 通过raise直接返回一个错误

@retry
def test_retry2():
    print("等待重试.....")
    return "hello" + 1  # 人为制造一个错误，这里我是把字符串和整数相加，因为类型不同，肯定会报错，所以会触发重试

上述2段代码运行后会一直打印“等待重试”，直至手工停止运行

例2【设置最大重试次数stop_after_attempt】

@retry(stop=stop_after_attempt(5))
def test_retry():
    print("等待重试.....")
    return "hello" + 1 

用 stop 接收 stop_after_attempt，当重试指定次数时，结束重试，如下重试了5次

例3【设置最大重时间，如果失败，则重试，一直重试5s】 

@retry(stop=stop_after_delay(5))
def test_retry():
    print("等待重试.....")
    return "hello" + 1 

例4【出现特定错误后重试】

@retry(retry=retry_if_exception_type(TypeError))
def test_retry1():
    print("等待重试.....")
    return "hello" + 1  # 捕获类型错误，当出现类型错误时重试

@retry(retry=retry_if_exception_type(SyntaxError))
def test_retry2():
    print("等待重试.....")
    raise SyntaxError  # 捕获语法错误，当出现语法错误时重试

例5【满足自定义的条件后重试】

# 首先定义了一个函数symbol，它的作用是判断传入的值是否为None；它返回一个布尔值，如果结果value=None，则返回true，否则返回False
def symbol(value):
    return value is None


# 装饰器中retry=retry_if_result(symbol)，表示把test_retry函数的结果传入symbol，判断test_retry的结果是否为None，
# 如果=None，就进行重试(retry),如果不等于None，就结束并返回函数值（所以达成重试的条件是test_retry的结果是否为条件函数定义的结果）
@retry(stop=stop_after_attempt(3), retry=retry_if_result(symbol), reraise=True)
def test_retry():
    print("等待重试.....")
    return None

symbol()函数是定义的条件函数，test_retry()函数是希望重试的函数，它俩通过装饰器中的retry_if_result()来关联，具体含义可以看下上述代码的注释

---

接下来开始处理我的接口测试脚本，用到是上面例5的自定义条件重试

首先处理需要重试的方法，我规定了当这个方法没有接收到推送过来的数据时，返回None

    def seal_regist(code):

        seal_data = self.get_seal_data(code)

        try:
            if seal_data["data"]["list"]:

                r = requests.post(url, json=payload, headers=headers)
                return r.json()

            else:
                print("列表中没有这条待用印数据{}，请检查系统～".format(code))
                return None

        except Exception as e:
            raise e

定义一个条件函数

def test_retry(value):
    """重试条件函数"""
    return value is None

给 seal_regist()函数加上retry装饰器

@retry(stop=stop_after_delay(10), retry=retry_if_result(test_retry))
    def seal_regist(code):
　　　　　　.....
　　　　　　.....

如果 seal_regist()返回None则重试，最大重试时间为10s

ps.因为重试函数中需要用到登陆cookie，之前是把登陆获取cookie的方法写到里面的，但是如果加上重试机制的话，当开始重试时会一直重新登录获取cookie，提示登陆频繁并导致登陆接口调用失败，所以为了避免这种情况，我把获取登陆cookie的方法放到了外面，这样无论重试几次都用开始获取到的一个cookie即可（所以如果有遇到和我类似情况的，把那些类似只需获取一次数据的方法放到外面，避免重复请求接口引发异常）