🔄重试机制

为了增强测试代码的健壮性,aomaker提供了一种重试机制,下面针对三种使用场景来做说明。

1.网络请求

aomakerBaseApi中,对发送http请求若遇到请求状态码大于400,会自动进行重试。

class BaseApi:
    IS_HTTP_RETRY = False
    HTTP_RETRY_COUNTS = 3
    HTTP_RETRY_INTERVAL = 2  # 单位:s

    def __init__(self):
        self.cache = Cache()
        self.config = Config().get_all()
        self._host = self.config.get('host')
        self._headers = self.cache.get('headers')
        self._response_callback = response_callback

这里有三个参数来控制http请求时的重试逻辑:

  • IS_HTTP_RETRY :总开关,默认关闭重试
  • HTTP_RETRY_COUNTS:重试次数,默认3次
  • HTTP_RETRY_INTERVAL:每次重试间隔的时间,默认2秒(单位秒)

那如何来修改这些参数呢? 前面说过,项目的所有ao都是继承于这个基类,所以可以继承该基类,实现一个项目的子类。

from aomaker.base.base_api imoort BaseApi


class MyBaseApi(BaseApi):
    IS_HTTP_RETRY = True
    HTTP_RETRY_COUNTS = 5
    HTTP_RETRY_INTERVAL = 3

然后所有的ao继承于该类即可,这样当所有http请求如果出现网络请求状态码>400后,将会每3秒重试一次,总共重试5次。

2.业务层面

除了在网络请求层面,还可以在业务层面加入重试,进一步加强健壮性。

比如,可以根据业务接口的业务码来做重试(大部分项目都会有), 假如这里有个项目的所有业务接口的响应都会返回一个ret_code,当这个code为0,代表业务正常,非0表示异常,那我希望当非0时,进行一定程度的重试,怎么做呢? 导入aomaker的重试模块:

from aomaker import aomaker


class Instance(BaseApi):

    @aomaker.retry(counts=2, interval=3, retry_condition=lambda resp: resp.get("ret_code") != 0)
    def get_instances(self):
        http_data = {
            "owner": self.owner,
            "action": "DescribeInstances",
            "reverse": 1,
            "status": ["pending", "running", "stopped", "suspended", "rescuing"],
        }
        resp = self.send_http(http_data)
        return resp

@aomaker.retry(counts=2,interval=3,retry_condition=lambda resp: resp.get("ret_code") != 0) ,解释下这个装饰器:

  • counts: 表示重试次数
  • interval:表示重试间隔
  • retry_condition:表示重试条件,这个参数接受一个bool值,为True就将进行重试,这里会提取被装饰函数的返回值来进行判断

除了这几个参数,该装饰器还有一个参数:exception_type 该参数可以指定特定异常,当触发该异常时,将会进行重试。

栗子:

@aomaker.retry(counts=2, interval=2, exception_type=TypeError)
def dem(num):
    if num==0:
        raise TypeError
    return num

dem(0)

3.代码片段

除了对函数、方法级别进行重试,还可以针对某段代码片段进行重试。

使用方法和上面的装饰器略有不同,看栗子: 先导入AoMakerRetry

from aomaker.aomaker import AoMakerRetry

def get_random_num():
    num = 6
    return num


def demo():
    print('start')
    for attempt in AoMakerRetry(counts=3, interval=2, exception_type=TypeError):
        with attempt:
            print('running...')
            random_ = get_random_num()
            if random_ == 6:
                raise TypeError
    print('end')


demo()

这里需要用for x in AoMakerRetry() with x: 这样固定的结构体,将需要重试的代码片段包裹起来,而AoMakerRetry类所接受的参数和上面装饰器是一样的。

上面代码的输出结果:

results matching ""

    No results matching ""