前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >python retry 和 retrying(失败重试)

python retry 和 retrying(失败重试)

作者头像
卓越笔记
发布2023-02-18 13:48:10
发布2023-02-18 13:48:10
1.2K00
代码可运行
举报
文章被收录于专栏:卓越笔记卓越笔记
运行总次数:0
代码可运行

retry

官:https://pypi.org/project/retry/

译:https://spaces.ac.cn/archives/3902

Easy to use retry decorator.

Features

  • No external dependency (stdlib only).
  • (Optionally) Preserve function signatures (pip install decorator).
  • Original traceback, easy to debug.

Installation

代码语言:javascript
代码运行次数:0
运行
复制
$ pip install retry

Examples

代码语言:javascript
代码运行次数:0
运行
复制
from retry import retry
代码语言:javascript
代码运行次数:0
运行
复制
@retry()
def make_trouble():
    '''Retry until succeed'''
代码语言:javascript
代码运行次数:0
运行
复制
@retry(ZeroDivisionError, tries=3, delay=2)
def make_trouble():
    '''Retry on ZeroDivisionError, raise error after 3 attempts, sleep 2 seconds between attempts.'''
代码语言:javascript
代码运行次数:0
运行
复制
@retry((ValueError, TypeError), delay=1, backoff=2)
def make_trouble():
    '''Retry on ValueError or TypeError, sleep 1, 2, 4, 8, ... seconds between attempts.'''
代码语言:javascript
代码运行次数:0
运行
复制
@retry((ValueError, TypeError), delay=1, backoff=2, max_delay=4)
def make_trouble():
    '''Retry on ValueError or TypeError, sleep 1, 2, 4, 4, ... seconds between attempts.'''
代码语言:javascript
代码运行次数:0
运行
复制
@retry(ValueError, delay=1, jitter=1)
def make_trouble():
    '''Retry on ValueError, sleep 1, 2, 3, 4, ... seconds between attempts.'''
代码语言:javascript
代码运行次数:0
运行
复制
# If you enable logging, you can get warnings like 'ValueError, retrying in
# 1 seconds'
if __name__ == '__main__':
    import logging
    logging.basicConfig()
    make_trouble()

retry_call

代码语言:javascript
代码运行次数:0
运行
复制
def retry_call(f, fargs=None, fkwargs=None, exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1,
               jitter=0,
               logger=logging_logger):
    """
    Calls a function and re-executes it if it failed.

    :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.
    """

This is very similar to the decorator, except that it takes a function and its arguments as parameters. The use case behind it is to be able to dynamically adjust the retry arguments.

代码语言:javascript
代码运行次数:0
运行
复制
import requests

from retry.api import retry_call


def make_trouble(service, info=None):
    if not info:
        info = ''
    r = requests.get(service + info)
    return r.text


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)

what_is_my_ip("conservative")

retrying

官:https://pypi.org/project/retrying/

译:https://www.jianshu.com/p/364377ffdcc1

Project description

Retrying is an Apache 2.0 licensed general-purpose retrying library, written in Python, to simplify the task of adding retry behavior to just about anything.

The simplest use case is retrying a flaky function whenever an Exception occurs until a value is returned.

代码语言:javascript
代码运行次数:0
运行
复制
import random
from retrying import retry

@retry
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("Broken sauce, everything is hosed!!!111one")
    else:
        return "Awesome sauce!"

print do_something_unreliable()

Features

  • Generic Decorator API
  • Specify stop condition (i.e. limit by number of attempts)
  • Specify wait condition (i.e. exponential backoff sleeping between attempts)
  • Customize retrying on Exceptions
  • Customize retrying on expected returned result

Installation

To install retrying, simply:

代码语言:javascript
代码运行次数:0
运行
复制
$ pip install retrying

Or, if you absolutely must:

代码语言:javascript
代码运行次数:0
运行
复制
$ easy_install retrying

But, you might regret that later.

Examples

As you saw above, the default behavior is to retry forever without waiting.

代码语言:javascript
代码运行次数:0
运行
复制
@retry
def never_give_up_never_surrender():
    print "Retry forever ignoring Exceptions, don't wait between retries"

Let’s be a little less persistent and set some boundaries, such as the number of attempts before giving up.

代码语言:javascript
代码运行次数:0
运行
复制
@retry(stop_max_attempt_number=7)
def stop_after_7_attempts():
    print "Stopping after 7 attempts"

We don’t have all day, so let’s set a boundary for how long we should be retrying stuff.

代码语言:javascript
代码运行次数:0
运行
复制
@retry(stop_max_delay=10000)
def stop_after_10_s():
    print "Stopping after 10 seconds"

Most things don’t like to be polled as fast as possible, so let’s just wait 2 seconds between retries.

代码语言:javascript
代码运行次数:0
运行
复制
@retry(wait_fixed=2000)
def wait_2_s():
    print "Wait 2 second between retries"

Some things perform best with a bit of randomness injected.

代码语言:javascript
代码运行次数:0
运行
复制
@retry(wait_random_min=1000, wait_random_max=2000)
def wait_random_1_to_2_s():
    print "Randomly wait 1 to 2 seconds between retries"

Then again, it’s hard to beat exponential backoff when retrying distributed services and other remote endpoints.

代码语言:javascript
代码运行次数:0
运行
复制
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)
def wait_exponential_1000():
    print "Wait 2^x * 1000 milliseconds between each retry, up to 10 seconds, then 10 seconds afterwards"

We have a few options for dealing with retries that raise specific or general exceptions, as in the cases here.

代码语言:javascript
代码运行次数:0
运行
复制
def retry_if_io_error(exception):
    """Return True if we should retry (in this case when it's an IOError), False otherwise"""
    return isinstance(exception, IOError)

@retry(retry_on_exception=retry_if_io_error)
def might_io_error():
    print "Retry forever with no wait if an IOError occurs, raise any other errors"

@retry(retry_on_exception=retry_if_io_error, wrap_exception=True)
def only_raise_retry_error_when_not_io_error():
    print "Retry forever with no wait if an IOError occurs, raise any other errors wrapped in RetryError"

We can also use the result of the function to alter the behavior of retrying.

代码语言:javascript
代码运行次数:0
运行
复制
def retry_if_result_none(result):
    """Return True if we should retry (in this case when result is None), False otherwise"""
    return result is None

@retry(retry_on_result=retry_if_result_none)
def might_return_none():
    print "Retry forever ignoring Exceptions with no wait if return value is None"

Any combination of stop, wait, etc. is also supported to give you the freedom to mix and match.

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2019-12-18 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • retry
    • Features
    • Installation
    • Examples
    • retry_call
  • retrying
    • Project description
    • Features
    • Installation
    • Examples
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档