昔我往矣

Python的重试模块retry

2020年07月9日

在某些场景下,代码执行出现异常的时候,我们希望代码能自动进行重试。比如,某个三方接口的token时效时间是30分钟,在编码的时候,我希望能主动尝试登录。如果接口返回401了,重新获取token,再用新token请求一次。
retry

Python中有个外部模块叫做retry,可以帮助我们解决这个重试的问题,retry的官方项目地址是 https://github.com/invl/retry/ 。使用pip进行安装:

$ pip install retry

一个不完整的简单demo

from retry import retry
import requests


class HttpError401(Exception):         # retry把Exception作为重试的条件,这里我们自定义HttpError401
    """ api return 401, Unauthorized """


@retry(HttpError401, tries=3, delay=1)   # 以装饰器的方式使用retry
def attempt_request():
    token = get_token()                  # 假定有一个获取token的方法
    headers = {
        "X-Token": token                 # 假定是通过http header发送token进行认证
    }
    r = requests.get(url, headers=headers)   # 假定有一个url
    if r.status_code == 401:            # 如果请求返回401,则执行下面的刷新token和抛出异常操作  
        renew_token()                      
        raise HttpError401
    return r.json()                    # 如果请求返回不是401,则获取请求的响应数据

根据注释,这段代码还是比较好理解的。下面说一下retry方法的参数:

  • HttpError401 是执行重试的条件,如果代码抛出这个异常就重试,可以定义多个异常,作为元组传入。
  • delay 是指重试的间隔时间,单位秒。
  • tries 是最大重试次数,默认-1,即无限重试,不达目的不罢休。

retry 模块还有其他很多有用的参数,比如backoff可以设置成倍的增加每次delay的时间间隔。max_delay可以设置最大的delay时间。

查看重试次数

如果我们想在被装饰函数中获取尝试的次数呢?据我所知,retry并没有提供相应的方法。那我们自己想办法吧。为了记录重试次数,我们需要引入额外的变量来记录执行次数,如下是一份完整的,但是没啥实际意义的代码:

from retry import retry
from random import randint
import logging


class NegativeError(Exception):
    """ Negative is not acceptable """


def random_divide():
    count = 0
    @retry((ZeroDivisionError, NegativeError), tries=5, delay=1, backoff=3)
    def _exec():
        nonlocal count       # Python3中引入的新关键字
        a = randint(-5, 5)
        b = randint(-2, 2)
        count += 1

        if a / b > 0:
            return f"get a positive number. after {count} times tried"
        if a / b < 0:
            raise NegativeError
    return _exec()


if __name__ == "__main__":
    logging.basicConfig()
    result = random_divide()
    print(result)

代码逻辑比较简单,不用太多解释。执行这段代码,你会得到如下的结果:

WARNING:retry.api:, retrying in 1 seconds...
WARNING:retry.api:, retrying in 3 seconds...
WARNING:retry.api:, retrying in 9 seconds...
get a positive number. after 4 times tried

在这个案例中,我们关注如下两点:

  1. 我们通过日志打印了retry的执行记录,结果显示第一次失败后,等待了1s开始第二次执行,第二次也失败后,等待了3s才开始执行,以此类推。这个delay时间是成倍数增加的,这个就是由retrybackoff参数决定的。
  2. 我们在被装饰的函数中获得了执行的次数, 也就是这里的count变量,在外层函数定义,通过nonlocal关键字,在内层函数中声明和修改。

retry是一个简单而且强大的Python库,更多用法可以到其官方仓库中学习。

当前暂无评论 »

添加新评论 »