我有一个循环,开头为for I,范围为(0,100)。正常情况下,它可以正常运行,但有时由于网络条件而出现故障。目前,我已经设置它,以便在失败时,它将继续在except子句中(继续到I的下一个数字)。
我是否可以将相同的数字重新分配给i,并再次运行失败的循环迭代?
我有一个循环,开头为for I,范围为(0,100)。正常情况下,它可以正常运行,但有时由于网络条件而出现故障。目前,我已经设置它,以便在失败时,它将继续在except子句中(继续到I的下一个数字)。
我是否可以将相同的数字重新分配给i,并再次运行失败的循环迭代?
当前回答
使用这个装饰器,您可以轻松地控制错误
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')
其他回答
如果你想要一个没有嵌套循环和成功调用break的解决方案,你可以为任何可迭代对象开发一个快速的可检索包装。这里有一个我经常遇到的网络问题的例子——保存的身份验证过期。它的用法是这样的:
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
我倾向于限制重试次数,这样如果某个特定项目出现问题,你就可以继续进行下一个项目,如下:
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.
我在我的代码中使用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
重新尝试的替代方案:坚韧和退缩(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页面上列出了重试。
这里有一个快速装饰器来处理这个问题。7行,没有依赖关系。
def retry(exception=Exception, retries=3, delay=0):
def wrap(func):
for i in range(retries):
try:
return func()
except exception as e:
print(f'Retrying {func.__name__}: {i}/{retries}')
time.sleep(delay)
raise e
return wrap
@retry()
def do_something():
...
@retry(HTTPError, retries=100, delay=3)
def download_something():
...
可以添加的一个功能是扩展异常以处理多个异常(splat一个列表)。