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

2022-03-15 16:57:07 浏览数 (1)

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

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

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

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

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

第1个解决方案

开始想到的解决方案是使用time.sleep()

当调用A接口后,等待一段时间,如 time.sleep(5),死等5s,然后再调用B接口

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

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

第2个解决方案

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

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

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

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

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

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

代码语言:javascript复制
@retry
def test_retry1():
    print("等待重试.....")
    raise Exception  # 通过raise直接返回一个错误

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

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

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

代码语言:javascript复制
@retry(stop=stop_after_attempt(5))
def test_retry():
    print("等待重试.....")
    return "hello"   1

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

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

代码语言:javascript复制
@retry(stop=stop_after_delay(5))
def test_retry():
    print("等待重试.....")
    return "hello"   1 

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

代码语言:javascript复制
@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【满足自定义的条件后重试】

代码语言:javascript复制
# 首先定义了一个函数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

代码语言:javascript复制
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

定义一个条件函数

代码语言:javascript复制
def test_retry(value):
    """重试条件函数"""
    return value is None

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

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

代码语言:javascript复制
@retry(stop=stop_after_delay(10), retry=retry_if_result(test_retry))
    def seal_regist(code):
      .....
      .....

ps.因为重试函数中需要用到登陆cookie,之前是把登陆获取cookie的方法写到里面的

但是如果加上重试机制的话,当开始重试时会一直重新登录获取cookie,提示登陆频繁并导致登陆接口调用失败

所以为了避免这种情况,我把获取登陆cookie的方法放到了外面,这样无论重试几次都用开始获取到的一个cookie即可(所以如果有遇到和我类似情况的,把那些类似只需获取一次数据的方法放到外面,避免重复请求接口引发异常)

0 人点赞