我有一个循环,开头为for I,范围为(0,100)。正常情况下,它可以正常运行,但有时由于网络条件而出现故障。目前,我已经设置它,以便在失败时,它将继续在except子句中(继续到I的下一个数字)。

我是否可以将相同的数字重新分配给i,并再次运行失败的循环迭代?


当前回答

如果您正在寻找的是重新尝试x次失败的尝试,那么单个for else循环可能就是您想要的。考虑这个例子,尝试了3次:

attempts = 3

for attempt in range(1, attempts+1):
    try:
        if attempt < 4:
            raise TypeError(f"Error raised on attempt: {attempt}")
        else:
            print(f'Attempt {attempt} finally worked.')
    except (TypeError) as error:
        print(f'Attempt {attempt} hit the exception.')
        continue
    else:
        break
else:
    print(f'Exit after final attempt: {attempt}')

print(f'\nGo on to execute other code ...')

给出输出:

Attempt 1 hit the exception.
Attempt 2 hit the exception.
Attempt 3 hit the exception.
Exit after final attempt: 3

Go on to execute other code ...

再试一次它就成功了

attempts = 4

给出输出:

Attempt 1 hit the exception.
Attempt 2 hit the exception.
Attempt 3 hit the exception.
Attempt 4 finally worked.

Go on to execute other code ...

其他回答

for _ in range(5):
    try:
        # replace this with something that may fail
        raise ValueError("foo")

    # replace Exception with a more specific exception
    except Exception as e:
        err = e
        continue

    # no exception, continue remainder of code
    else:
        break

# did not break the for loop, therefore all attempts
# raised an exception
else:
    raise err

我的版本与上面的几个类似,但没有使用单独的while循环,如果所有重试都失败,则重新引发最新的异常。可以显式地在顶部设置err = None,但不是严格必要的,因为它只应该在出现错误时执行最后一个else块,因此设置了err。

重新尝试的替代方案:坚韧和退缩(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页面上列出了重试。

不使用那些丑陋的while循环的更“功能性”的方法:

def tryAgain(retries=0):
    if retries > 10: return
    try:
        # Do stuff
    except:
        retries+=1
        tryAgain(retries)

tryAgain()

使用递归

for i in range(100):
    def do():
        try:
            ## Network related scripts
        except SpecificException as ex:
            do()
    do() ## invoke do() whenever required inside this loop

以下是我对这个问题的看法。下面的重试功能支持以下特性:

当调用成功时返回被调用函数的值 如果尝试失败,则引发被调用函数的异常 尝试次数限制(0表示无限) 在尝试之间等待(线性或指数) 仅当异常是特定异常类型的实例时重试。 可选的尝试记录

import time

def retry(func, ex_type=Exception, limit=0, wait_ms=100, wait_increase_ratio=2, logger=None):
    attempt = 1
    while True:
        try:
            return func()
        except Exception as ex:
            if not isinstance(ex, ex_type):
                raise ex
            if 0 < limit <= attempt:
                if logger:
                    logger.warning("no more attempts")
                raise ex

            if logger:
                logger.error("failed execution attempt #%d", attempt, exc_info=ex)

            attempt += 1
            if logger:
                logger.info("waiting %d ms before attempt #%d", wait_ms, attempt)
            time.sleep(wait_ms / 1000)
            wait_ms *= wait_increase_ratio

用法:

def fail_randomly():
    y = random.randint(0, 10)
    if y < 10:
        y = 0
    return x / y


logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(stream=sys.stdout))

logger.info("starting")
result = retry.retry(fail_randomly, ex_type=ZeroDivisionError, limit=20, logger=logger)
logger.info("result is: %s", result)

更多信息请看我的帖子。