python简单的retry模块

背景

我们经常遇到一个场景，就是如果操作失败则需要多次重试某些操作，这种情况下，如果想优雅的实现功能，又不关心重试逻辑，则可以学习该模块

安装

1
pip install retry

API 介绍

retry decorator

1
2
3
4
5
6
7
8
9
10
11
12
13
def retry(exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=logging_logger):
    """Return a retry decorator.

    :param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
    :param tries: the maximum number of attempts. default: -1 (infinite).
    :param delay: initial delay between attempts. default: 0.
    :param max_delay: the maximum value of delay. default: None (no limit).
    :param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
    :param jitter: extra seconds added to delay between attempts. default: 0.
                   fixed if a number, random if a range tuple (min, max)
    :param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
                   default: retry.logging_logger. if None, logging is disabled.
    """

retry 参数介绍

  • 无参
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from retry import retry

@retry()
def make_trouble():
    '''Retry until succeed'''
    print ('retrying...')
    raise

if __name__ == '__main__':
    make_trouble()

# 输出: 一直重试，直到运行成功
retrying...
retrying...
retrying...
retrying...
retrying...
retrying...
  • Exception参数， 默认 Exception, 只捕获重试指定的异常，可以是元组
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@retry(ZeroDivisionError, tries=3, delay=2)
def make_trouble():
    '''Retry on ZeroDivisionError, raise error after 3 attempts, sleep 2 seconds between attempts.'''
    print 'aaa'
    a = 1/0

if __name__ == '__main__':
    make_trouble()

输出:
aaa
aaa
Traceback (most recent call last):
  File "E:/WORKSPACE/document/document/test/1.py", line 20, in <module>
    make_trouble()
  File "<decorator-gen-2>", line 2, in make_trouble
  File "D:\python27\WinPython-64bit-2.7.10.3\python-2.7.10.amd64\lib\site-packages\retry\api.py", line 74, in retry_decorator
    logger)
  File "D:\python27\WinPython-64bit-2.7.10.3\python-2.7.10.amd64\lib\site-packages\retry\api.py", line 33, in __retry_internal
    return f()
  File "E:/WORKSPACE/document/document/test/1.py", line 16, in make_trouble
    a = 1/0
ZeroDivisionError: integer division or modulo by zero
aaa
  • backoff参数，尝试间隔时间，成倍数增加
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import time
@retry((ValueError, TypeError), delay=1, backoff=2)
def make_trouble():
    '''Retry on ValueError or TypeError, sleep 1, 2, 4, 8, ... seconds between attempts.'''
    print (1,  int(time.time()))

    raise ValueError('a')

if __name__ == '__main__':
    make_trouble()

输出:
(1, 1504107288)
(1, 1504107289)
(1, 1504107291)
(1, 1504107295)
(1, 1504107303)
(1, 1504107319)
(1, 1504107351)
  • max_delay 指定最大间隔时间，backoff参数触发的休眠时间大于max_delay时，休眠时间以max_delay为准则
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import time

@retry((ValueError, TypeError), delay=1, backoff=2, max_delay=8)
def make_trouble():
    '''Retry on ValueError or TypeError, sleep 1, 2, 4, 4, ... seconds between attempts.'''

    print (1, int(time.time()))
    raise ValueError('aa')


if __name__ == '__main__':
    make_trouble()

输出：
(1, 1504107496)
(1, 1504107497)
(1, 1504107499)
(1, 1504107503)
(1, 1504107511)
(1, 1504107519)
(1, 1504107527)
(1, 1504107535)
  • jitter参数，累加，以及异常触发的日志
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import time

@retry(ValueError, delay=1, jitter=1)
def make_trouble():
    '''Retry on ValueError, sleep 1, 2, 3, 4, ... seconds between attempts.'''
    print (1, int(time.time()))
    raise ValueError('e')

if __name__ == '__main__':
    import logging
    logging.basicConfig()
    make_trouble()

输出：
WARNING:retry.api:e, retrying in 1 seconds...
(1, 1504107644)
WARNING:retry.api:e, retrying in 2 seconds...
(1, 1504107645)
WARNING:retry.api:e, retrying in 3 seconds...
(1, 1504107647)
WARNING:retry.api:e, retrying in 4 seconds...
(1, 1504107650)
WARNING:retry.api:e, retrying in 5 seconds...
(1, 1504107654)
WARNING:retry.api:e, retrying in 6 seconds...
(1, 1504107659)
(1, 1504107665)
WARNING:retry.api:e, retrying in 7 seconds...
(1, 1504107672)
WARNING:retry.api:e, retrying in 8 seconds...

retry_call

def retry_call(f, fargs=None, fkwargs=None, exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1,
jitter=0,
logger=logging_logger):

“””

调用函数并在失败时重新执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
:param f: the function to execute.
:param fargs: the positional arguments of the function to execute.
:param fkwargs: the named arguments of the function to execute.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
               fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
               default: retry.logging_logger. if None, logging is disabled.
:returns: the result of the f function.
"""

例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import requests
from retry.api import retry_call

def make_trouble(service, info=None):
    if not info:
        info = ''
    print ('retry..., service: {},  info: {}'.format(service, info))
    r = requests.get(service + info)
    print r.text
    raise Exception('info')

def what_is_my_ip(approach=None):
    if approach == "optimistic":
        tries = 1
    elif approach == "conservative":
        tries = 3
    else:
        # skeptical
        tries = -1
    result = retry_call(make_trouble, fargs=["http://ipinfo.io/"], fkwargs={"info": "ip"}, tries=tries)
    print(result)

if __name__ == '__main__':
    import logging
    logging.basicConfig()
    what_is_my_ip("conservative")


输出:
retry..., service: http://ipinfo.io/,  info: ip
118.113.1.255

retry..., service: http://ipinfo.io/,  info: ip
WARNING:retry.api:info, retrying in 0 seconds...
WARNING:retry.api:info, retrying in 0 seconds...
118.113.1.255

retry..., service: http://ipinfo.io/,  info: ip
Traceback (most recent call last):
  File "E:/WORKSPACE/document/document/test/1.py", line 74, in <module>
    what_is_my_ip("conservative")
  File "E:/WORKSPACE/document/document/test/1.py", line 66, in what_is_my_ip
    result = retry_call(make_trouble, fargs=["http://ipinfo.io/"], fkwargs={"info": "ip"}, tries=tries)
  File "D:\python27\WinPython-64bit-2.7.10.3\python-2.7.10.amd64\lib\site-packages\retry\api.py", line 101, in retry_call
    return __retry_internal(partial(f, *args, **kwargs), exceptions, tries, delay, max_delay, backoff, jitter, logger)
  File "D:\python27\WinPython-64bit-2.7.10.3\python-2.7.10.amd64\lib\site-packages\retry\api.py", line 33, in __retry_internal
    return f()
  File "E:/WORKSPACE/document/document/test/1.py", line 54, in make_trouble
    raise Exception('info')
Exception: info
118.113.1.255

其他参数同retry decorator
官网

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码
× 文章目录
  1. 1. 背景
  2. 2. 安装
  3. 3. API 介绍
    1. 3.1. retry decorator
    2. 3.2. retry 参数介绍
    3. 3.3. retry_call
    4. 3.4. 例子

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

你好<br><br>我是帅比萌木盖，爱搞机<br>欢迎观看<br>如想联系：grmg@foxmail.com