重新尝试的替代方案:坚韧和退缩(2020年更新)

重新尝试库是以前的方法，但遗憾的是，它有一些bug，自2016年以来就没有任何更新。其他的选择似乎是后退和坚韧。在写这篇文章的时候，tenacity有更多的GItHub星(2.3k vs 1.2k)，并且最近更新了，因此我选择使用它。这里有一个例子:

from functools import partial
import random # producing random errors for this example

from tenacity import retry, stop_after_delay, wait_fixed, retry_if_exception_type

# Custom error type for this example
class CommunicationError(Exception):
    pass

# Define shorthand decorator for the used settings.
retry_on_communication_error = partial(
    retry,
    stop=stop_after_delay(10),  # max. 10 seconds wait.
    wait=wait_fixed(0.4),  # wait 400ms 
    retry=retry_if_exception_type(CommunicationError),
)()


@retry_on_communication_error
def do_something_unreliable(i):
    if random.randint(1, 5) == 3:
        print('Run#', i, 'Error occured. Retrying.')
        raise CommunicationError()

for i in range(100):
    do_something_unreliable(i)

上面的代码输出如下:

Run# 3 Error occured. Retrying.
Run# 5 Error occured. Retrying.
Run# 6 Error occured. Retrying.
Run# 6 Error occured. Retrying.
Run# 10 Error occured. Retrying.
.
.
.

坚韧的更多设置。坚韧GitHub页面上列出了重试。

我在我的代码中使用following，

   for i in range(0, 10):
    try:
        #things I need to do
    except ValueError:
        print("Try #{} failed with ValueError: Sleeping for 2 secs before next try:".format(i))
        time.sleep(2)
        continue
    break

我最近用我的python解决了这个问题，我很高兴与stackoverflow的访问者分享，如果需要请给予反馈。

print("\nmonthly salary per day and year converter".title())
print('==' * 25)


def income_counter(day, salary, month):
    global result2, result, is_ready, result3
    result = salary / month
    result2 = result * day
    result3 = salary * 12
    is_ready = True
    return result, result2, result3, is_ready


i = 0
for i in range(5):
    try:
        month = int(input("\ntotal days of the current month: "))
        salary = int(input("total salary per month: "))
        day = int(input("Total Days to calculate> "))
        income_counter(day=day, salary=salary, month=month)
        if is_ready:
            print(f'Your Salary per one day is: {round(result)}')
            print(f'your income in {day} days will be: {round(result2)}')
            print(f'your total income in one year will be: {round(result3)}')
            break
        else:
            continue
    except ZeroDivisionError:
        is_ready = False
        i += 1
        print("a month does'nt have 0 days, please try again")
        print(f'total chances left: {5 - i}')
    except ValueError:
        is_ready = False
        i += 1
        print("Invalid value, please type a number")
        print(f'total chances left: {5 - i}')

Decorator是一个很好的方法。

from functools import wraps
import time

class retry:
    def __init__(self, success=lambda r:True, times=3, delay=1, raiseexception=True, echo=True):
        self.success = success
        self.times = times
        self.raiseexception = raiseexception
        self.echo = echo
        self.delay = delay
    def retry(fun, *args, success=lambda r:True, times=3, delay=1, raiseexception=True, echo=True, **kwargs):
        ex = Exception(f"{fun} failed.")
        r = None
        for i in range(times):
            if i > 0:
                time.sleep(delay*2**(i-1))
            try:
                r = fun(*args, **kwargs)
                s = success(r)
            except Exception as e:
                s = False
                ex = e
                # raise e
            if not s:
                continue
            return r
        else:
            if echo:
                print(f"{fun} failed.", "args:", args, kwargs, "\nresult: %s"%r)
            if raiseexception:
                raise ex
    def __call__(self, fun):
        @wraps(fun)
        def wraper(*args, retry=0, **kwargs):
            retry = retry if retry>0 else self.times
            return self.__class__.retry(fun, *args, 
                                        success=self.success, 
                                        times=retry,
                                        delay=self.delay,
                                        raiseexception = self.raiseexception,
                                        echo = self.echo,
                                        **kwargs)
        return wraper

一些用法示例:

@retry(success=lambda x:x>3, times=4, delay=0.1)
def rf1(x=[]):
    x.append(1)
    print(x)
    return len(x)
> rf1()

[1]
[1, 1]
[1, 1, 1]
[1, 1, 1, 1]

4

@retry(success=lambda x:x>3, times=4, delay=0.1)
def rf2(l=[], v=1):
    l.append(v)
    print(l)
    assert len(l)>4
    return len(l)
> rf2(v=2, retry=10) #overwite times=4

[2]
[2, 2]
[2, 2, 2]
[2, 2, 2, 2]
[2, 2, 2, 2, 2]

5

> retry.retry(lambda a,b:a+b, 1, 2, times=2)

3

> retry.retry(lambda a,b:a+b, 1, "2", times=2)

TypeError: unsupported operand type(s) for +: 'int' and 'str'

带超时的通用解决方案:

import time

def onerror_retry(exception, callback, timeout=2, timedelta=.1):
    end_time = time.time() + timeout
    while True:
        try:
            yield callback()
            break
        except exception:
            if time.time() > end_time:
                raise
            elif timedelta > 0:
                time.sleep(timedelta)

用法:

for retry in onerror_retry(SomeSpecificException, do_stuff):
    retry()

我喜欢laurent-laporte的回答。下面是我的版本，它包装在一个类与静态方法和一些例子。我实现了重试计数作为另一种重试方式。还增加了kwargs。

from typing import List
import time


class Retry:
    @staticmethod
    def onerror_retry(exception, callback, retries: int = 0, timeout: float = 0, timedelta: float = 0,
                      errors: List = None, **kwargs):
        """

        @param exception: The exception to trigger retry handling with.
        @param callback: The function that will potentially fail with an exception
        @param retries: Optional total number of retries, regardless of timing if this threshold is met, the call will
                        raise the exception.
        @param timeout: Optional total amount of time to do retries after which the call will raise an exception
        @param timedelta: Optional amount of time to sleep in between calls
        @param errors: A list to receive all the exceptions that were caught.
        @param kwargs: An optional key value parameters to pass to the function to retry.
        """
        for retry in Retry.__onerror_retry(exception, callback, retries, timeout, timedelta, errors, **kwargs):
            if retry: retry(**kwargs)  # retry will be None when all retries fail.

    @staticmethod
    def __onerror_retry(exception, callback, retries: int = 0, timeout: float = 0, timedelta: float = 0,
                        errors: List = None, **kwargs):
        end_time = time.time() + timeout
        continues = 0
        while True:
            try:
                yield callback(**kwargs)
                break
            except exception as ex:
                print(ex)
                if errors:
                    errors.append(ex)

                continues += 1
                if 0 < retries < continues:
                    print('ran out of retries')
                    raise

                if timeout > 0 and time.time() > end_time:
                    print('ran out of time')
                    raise
                elif timedelta > 0:
                    time.sleep(timedelta)


err = 0

#
# sample dumb fail function
def fail_many_times(**kwargs):
    global err
    err += 1
    max_errors = kwargs.pop('max_errors', '') or 1
    if err < max_errors:
        raise ValueError("I made boo boo.")
    print("Successfully did something.")

#
# Example calls
try:
    #
    # retries with a parameter that overrides retries... just because
    Retry.onerror_retry(ValueError, fail_many_times, retries=5, max_errors=3)
    err = 0
    #
    # retries that run out of time, with 1 second sleep between retries.
    Retry.onerror_retry(ValueError, fail_many_times, timeout=5, timedelta=1, max_errors=30)
except Exception as err:
    print(err)