我有一个以for i in range(0, 100)
. 通常它可以正常运行,但有时会由于网络状况而失败。目前我已经设置了它,以便在失败时,它将continue
在 except 子句中(继续到下一个数字i
)。
我是否可以重新分配相同的数字i
并再次运行循环的失败迭代?
我有一个以for i in range(0, 100)
. 通常它可以正常运行,但有时会由于网络状况而失败。目前我已经设置了它,以便在失败时,它将continue
在 except 子句中(继续到下一个数字i
)。
我是否可以重新分配相同的数字i
并再次运行循环的失败迭代?
在你的 for 循环中做一个while True
,把你的代码放在里面,只有当你的代码成功时才try
从那个循环中中断。while
for i in range(0,100):
while True:
try:
# do stuff
except SomeSpecificException:
continue
break
我更喜欢限制重试次数,这样如果该特定项目出现问题,您最终将继续进行下一个,因此:
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.
更新 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")
这是一个与其他解决方案类似的解决方案,但如果在规定的次数或重试中未成功,则会引发异常。
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
retrying
:tenacity
和backoff
(2020 更新)的替代方案retrying库以前是要走的路,但遗憾的是它有一些错误,自 2016 年以来没有任何更新。其他选择似乎是backoff和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 页面上tenacity.retry
列出了更多设置。
不使用那些丑陋的 while 循环的更“实用”的方法:
def tryAgain(retries=0):
if retries > 10: return
try:
# Do stuff
except:
retries+=1
tryAgain(retries)
tryAgain()
最明确的方法是显式设置i
. 例如:
i = 0
while i < 100:
i += 1
try:
# do stuff
except MyException:
continue
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
设置时才执行最后一个块。
具有超时的通用解决方案:
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()
使用 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
使用递归
for i in range(100):
def do():
try:
## Network related scripts
except SpecificException as ex:
do()
do() ## invoke do() whenever required inside this loop
装饰器是一个很好的方法。
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'
您可以使用 Python 重试包。 重试
它是用 Python 编写的,以简化向几乎任何事情添加重试行为的任务。
我在我的代码中使用以下内容,
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
attempts = 3
while attempts:
try:
...
...
<status ok>
break
except:
attempts -=1
else: # executed only break was not raised
<status failed>
如果您想要一个没有嵌套循环并调用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
这是我对这个问题的看法。以下retry
函数支持以下功能:
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)
有关更多信息,请参阅我的帖子。
我喜欢为此使用布尔值,如下所示:
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
使用这个装饰器,您可以轻松控制错误
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')
我使用它,它可以用于任何功能:
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)
如果重试失败的尝试 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 ...
我最近与我的 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}')
这是我关于如何解决这个问题的想法:
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)
仅当 try 子句成功时才增加循环变量