我有一个循环,开头为for I,范围为(0,100)。正常情况下,它可以正常运行,但有时由于网络条件而出现故障。目前,我已经设置它,以便在失败时,它将继续在except子句中(继续到I的下一个数字)。
我是否可以将相同的数字重新分配给i,并再次运行失败的循环迭代?
我有一个循环,开头为for I,范围为(0,100)。正常情况下,它可以正常运行,但有时由于网络条件而出现故障。目前,我已经设置它,以便在失败时,它将继续在except子句中(继续到I的下一个数字)。
我是否可以将相同的数字重新分配给i,并再次运行失败的循环迭代?
当前回答
在Python装饰器库中也有类似的东西。
请记住,它不测试异常,而是测试返回值。它会重新尝试,直到被修饰的函数返回True。
稍微修改一下版本就可以了。
其他回答
更新2021-12-01:
自2016年6月起,不再维护重试包。 考虑使用活动的fork github.com/jd/tenacity,或者github.com/litl/backoff。
重试包是在失败时重试代码块的好方法。
例如:
@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")
不使用那些丑陋的while循环的更“功能性”的方法:
def tryAgain(retries=0):
if retries > 10: return
try:
# Do stuff
except:
retries+=1
tryAgain(retries)
tryAgain()
这里有一个快速装饰器来处理这个问题。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一个列表)。
Decorator是一个很好的方法。
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'
尝试次数= 3 而尝试: 试一试: ... ... <状态好了> 打破 除了: 尝试- = 1 Else: #已执行的唯一break未被引发 < >失败状态