346

我有一个以for i in range(0, 100). 通常它可以正常运行,但有时会由于网络状况而失败。目前我已经设置了它,以便在失败时,它将continue在 except 子句中(继续到下一个数字i)。

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

4

25 回答 25

476

在你的 for 循环中做一个while True,把你的代码放在里面,只有当你的代码成功时才try从那个循环中中断。while

for i in range(0,100):
    while True:
        try:
            # do stuff
        except SomeSpecificException:
            continue
        break
于 2010-01-18T05:02:20.820 回答
282

我更喜欢限制重试次数,这样如果该特定项目出现问题,您最终将继续进行下一个,因此:

for i in range(100):
  for attempt in range(10):
    try:
      # do thing
    except:
      # perhaps reconnect, etc.
    else:
      break
  else:
    # we failed all the attempts - deal with the consequences.
于 2011-10-05T15:03:37.973 回答
96

更新 2021-12-01

自 2016 年 6 月起,不再维护重试包。考虑使用活动分支github.com/jd/tenacity,或者github.com/litl/backoff


retrying是在失败时重试代码块的好方法。

例如:

@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")
于 2014-09-18T12:29:43.597 回答
35

这是一个与其他解决方案类似的解决方案,但如果在规定的次数或重试中未成功,则会引发异常。

tries = 3
for i in range(tries):
    try:
        do_the_thing()
    except KeyError as e:
        if i < tries - 1: # i is zero indexed
            continue
        else:
            raise
    break
于 2016-01-19T20:01:18.800 回答
19

retrying:tenacitybackoff(2020 更新)的替代方案

retrying库以前是要走的路,但遗憾的是它有一些错误,自 2016 年以来没有任何更新。其他选择似乎是backofftenacity。在写这篇文章的时候,坚韧有更多的 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 页面上tenacity.retry列出了更多设置。

于 2020-06-01T12:40:20.993 回答
15

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

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

tryAgain()
于 2010-10-28T20:46:42.283 回答
11

最明确的方法是显式设置i. 例如:

i = 0
while i < 100:
    i += 1
    try:
        # do stuff

    except MyException:
        continue
于 2010-01-18T05:13:47.440 回答
9
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设置时才执行最后一个块。

于 2019-02-05T15:50:25.207 回答
6

具有超时的通用解决方案:

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()
于 2014-12-04T17:26:13.200 回答
4

Python 装饰器库中有类似的东西。

请记住,它不测试异常,而是返回值。它会重试,直到修饰函数返回 True。

稍作修改的版本应该可以解决问题。

于 2011-06-15T07:58:16.273 回答
4

使用 while 和计数器:

count = 1
while count <= 3:  # try 3 times
    try:
        # do_the_logic()
        break
    except SomeSpecificException as e:
        # If trying 3rd time and still error?? 
        # Just throw the error- we don't have anything to hide :)
        if count == 3:
            raise
        count += 1
于 2016-08-31T11:57:34.353 回答
4

使用递归

for i in range(100):
    def do():
        try:
            ## Network related scripts
        except SpecificException as ex:
            do()
    do() ## invoke do() whenever required inside this loop
于 2016-08-31T12:06:23.130 回答
4

装饰器是一个很好的方法。

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'
于 2021-04-23T08:30:31.700 回答
3

您可以使用 Python 重试包。 重试

它是用 Python 编写的,以简化向几乎任何事情添加重试行为的任务。

于 2017-10-18T17:13:08.953 回答
2

我在我的代码中使用以下内容,

   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
于 2018-11-21T07:19:11.220 回答
2

attempts = 3
while attempts:
  try:
     ...
     ...
     <status ok>
     break
  except:
    attempts -=1
else: # executed only break was not  raised
   <status failed>

于 2020-04-21T08:59:01.553 回答
1

如果您想要一个没有嵌套循环并调用break成功的解决方案,您可以为任何可迭代开发快速包装retriable。这是我经常遇到的网络问题的示例 - 保存的身份验证过期。它的用法如下:

client = get_client()
smart_loop = retriable(list_of_values):

for value in smart_loop:
    try:
        client.do_something_with(value)
    except ClientAuthExpired:
        client = get_client()
        smart_loop.retry()
        continue
    except NetworkTimeout:
        smart_loop.retry()
        continue
于 2018-07-24T21:28:48.000 回答
1

这是我对这个问题的看法。以下retry函数支持以下功能:

  • 成功时返回被调用函数的值
  • 如果尝试用尽,则引发调用函数的异常
  • 尝试次数限制(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)

有关更多信息,请参阅我的帖子

于 2020-05-25T15:45:23.337 回答
0

我喜欢为此使用布尔值,如下所示:

success = False
num_try = 0
while success is False:
    if num_try >= 10: # or any number
        # handle error how  you please
    try:
        # code
        success = True
    except Exception as e:
        # record or do something with exception if needed
        num_try += 1
于 2021-07-05T05:18:15.250 回答
0

使用这个装饰器,您可以轻松控制错误

class catch:
    def __init__(self, max=1, callback=None):
        self.max = max 
        self.callback = callback 
    
    def set_max(self, max):
        self.max = max
    
    def handler(self, *args, **kwargs):
        self.index = 0
        while self.index < self.max: 
            self.index += 1
            try:
                self.func(self, *args, **kwargs)
        
            except Exception as error:
                if callable(self.callback):
                    self.callback(self, error, args, kwargs)
                
    def __call__(self, func):
        self.func = func
        return self.handler

import time
def callback(cls, error, args, kwargs):
    print('func args', args, 'func kwargs', kwargs)
    print('error', repr(error), 'trying', cls.index)
    if cls.index == 2:
        cls.set_max(4)
    
    else:
        time.sleep(1)
    
    
@catch(max=2, callback=callback)  
def test(cls, ok, **kwargs):
    raise ValueError('ok')

test(1, message='hello')
于 2021-09-26T12:39:41.053 回答
0

我使用它,它可以用于任何功能:

def run_with_retry(func: callable, max_retries: int = 3, wait_seconds: int = 2, **func_params):
num_retries = 1
while True:
    try:
        return func(*func_params.values())
    except Exception as e:
        if num_retries > max_retries:
            print('we have reached maximum errors and raising the exception')
            raise e
        else:
            print(f'{num_retries}/{max_retries}')
            print("Retrying error:", e)
            num_retries += 1
            sleep(wait_seconds)

像这样调用:

    def add(val1, val2):
        return val1 + val2

    run_with_retry(func=add, param1=10, param2=20)
于 2021-10-13T13:58:59.283 回答
0

如果重试失败的尝试 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 ...
于 2022-02-16T16:41:49.917 回答
-2

我最近与我的 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}')
于 2019-09-16T02:05:46.077 回答
-3

这是我关于如何解决这个问题的想法:

j = 19
def calc(y):
    global j
    try:
        j = j + 8 - y
        x = int(y/j)   # this will eventually raise DIV/0 when j=0
        print("i = ", str(y), " j = ", str(j), " x = ", str(x))
    except:
        j = j + 1   # when the exception happens, increment "j" and retry
        calc(y)
for i in range(50):
    calc(i)
于 2015-12-24T17:35:50.750 回答
-10

仅当 try 子句成功时才增加循环变量

于 2010-01-18T09:34:27.510 回答