我有一个循环，开头为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
我喜欢使用bool值，如下所示:
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')
重新尝试的替代方案:坚韧和退缩(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页面上列出了重试。
你可以有一个专门的函数使用返回短路结果。比如这样:
def my_function_with_retries(..., max_retries=100):
for attempt in range(max_retries):
try:
return my_function(...)
except SomeSpecificException as error:
logging.warning(f"Retrying after failed execution: {error}")
raise SomeOtherException()
