在 Python 中尝试,直到没有错误为止

我在 Python 中有一段代码,它似乎在概率上导致了一个错误,因为它正在访问一个服务器,有时该服务器有500个内部服务器错误。我想继续尝试,直到我没有得到错误。我的解决办法是:

while True:
try:
#code with possible error
except:
continue
else:
#the rest of the code
break

This seems like a hack to me. Is there a more Pythonic way to do this?

163518 次浏览

也许是这样的:

connected = False


while not connected:
try:
try_connect()
connected = True
except ...:
pass

不会再干净了。这不是一件干净的事。在最好的情况下(因为 break的条件与 while一致,所以无论如何这将更具可读性) ,您可以创建一个变量 result = None并在它 is None的时候循环。你也应该调整变量,你可以用语义上可能正确的 pass代替 continue(你不在乎是否发生错误,你只是想忽略它) ,然后删除 break-这也会让剩下的代码,只执行一次,从循环中出来。还要注意,由于 文件所提供的资料的原因,裸 except:子句是有害的。

综合以上所有内容的例子:

result = None
while result is None:
try:
# connect
result = get_data(...)
except:
pass
# other code that uses result but is not involved in getting it

也许是室内设计师? 您可以将希望重试的异常列表和/或尝试次数作为装饰器参数传递。

def retry(exceptions=None, tries=None):
if exceptions:
exceptions = tuple(exceptions)
def wrapper(fun):
def retry_calls(*args, **kwargs):
if tries:
for _ in xrange(tries):
try:
fun(*args, **kwargs)
except exceptions:
pass
else:
break
else:
while True:
try:
fun(*args, **kwargs)
except exceptions:
pass
else:
break
return retry_calls
return wrapper




from random import randint


@retry([NameError, ValueError])
def foo():
if randint(0, 1):
raise NameError('FAIL!')
print 'Success'


@retry([ValueError], 2)
def bar():
if randint(0, 1):
raise ValueError('FAIL!')
print 'Success'


@retry([ValueError], 2)
def baz():
while True:
raise ValueError('FAIL!')


foo()
bar()
baz()

of course the 'try' part should be moved to another funcion becouse we using it in both loops but it's just example;)

Here is a short piece of code I use to capture the error as a string. Will retry till it succeeds. This catches all exceptions but you can change this as you wish.

start = 0
str_error = "Not executed yet."
while str_error:
try:
# replace line below with your logic , i.e. time out, max attempts
start = raw_input("enter a number, 0 for fail, last was {0}: ".format(start))
new_val = 5/int(start)
str_error=None
except Exception as str_error:
pass

警告: 此代码将陷入永久循环,直到没有异常发生。这只是一个简单的例子,可能需要您更快地打破循环,或者在重试之间休眠。

下面是一个在4次尝试之后都会失败的故障,并且在每次尝试之间等待2秒钟。按照你希望得到的改变,从这个例子中得到你想要的:

from time import sleep


for x in range(0, 4):  # try 4 times
try:
# msg.send()
# put your logic here
str_error = None
except Exception as str_error:
pass


if str_error:
sleep(2)  # wait for 2 seconds before trying to fetch the data again
else:
break

这里有一个后退的例子:

from time import sleep


sleep_time = 2
num_retries = 4
for x in range(0, num_retries):
try:
# put your logic here
str_error = None
except Exception as e:
str_error = str(e)


if str_error:
sleep(sleep_time)  # wait before trying to fetch the data again
sleep_time *= 2  # Implement your backoff algorithm here i.e. exponential backoff
else:
break

下面是我编写的一个实用函数,它将重试包装成一个更整洁的包。它使用相同的基本结构,但防止重复。可以对其进行修改,以便在最后一次尝试时相对容易地捕获和重新抛出异常。

def try_until(func, max_tries, sleep_time):
for _ in range(0,max_tries):
try:
return func()
except:
sleep(sleep_time)
raise WellNamedException()
#could be 'return sensibleDefaultValue'

可以这样叫

result = try_until(my_function, 100, 1000)

如果需要向 my_function传递参数,可以通过让 try_until转发参数,或者将其包装为无参数 lambda:

result = try_until(lambda : my_function(x,y,z), 100, 1000)

itertools.iter_except菜谱封装了“重复调用函数直到引发异常”的思想。它类似于公认的答案,但是配方提供了一个迭代器。

食谱如下:

def iter_except(func, exception, first=None):
""" Call a function repeatedly until an exception is raised."""
try:
if first is not None:
yield first()            # For database APIs needing an initial cast to db.first()
while True:
yield func()
except exception:
pass

您当然可以直接实现后面的代码。为了方便起见,我使用了一个单独的库 more_itertools,它为我们实现了这个菜谱(可选)。

密码

import more_itertools as mit


list(mit.iter_except([0, 1, 2].pop, IndexError))
# [2, 1, 0]

细节

在这里,对列表对象的每次迭代调用 pop方法(或给定函数) ,直到引发 IndexError

对于您的情况,给定一些 connect_function和预期的错误,您可以创建一个迭代器来重复调用函数,直到引发异常,例如。

mit.iter_except(connect_function, ConnectionError)

At this point, treat it as any other iterator by looping over it or calling next().

像大多数其他人一样,我建议尝试有限的次数,并在尝试之间睡觉。这样,您就不会发现自己处于一个无限循环中,以防远程服务器发生什么事情。

我还建议只有当您得到所期望的特定异常时才继续。这样,您仍然可以处理您可能没有预料到的异常。

from urllib.error import HTTPError
import traceback
from time import sleep




attempts = 10
while attempts > 0:
try:
#code with possible error
except HTTPError:
attempts -= 1
sleep(1)
continue
except:
print(traceback.format_exc())


#the rest of the code
break

而且,您不需要 else 块。由于“除外”块中的“继续”,您将跳过循环的其余部分,直到 try 块工作、 while 条件得到满足或出现 HTTPError 以外的异常为止。

e = ''
while e == '':
try:
response = ur.urlopen('https://https://raw.githubusercontent.com/MrMe42/Joe-Bot-Home-Assistant/mac/Joe.py')
e = ' '
except:
print('Connection refused. Retrying...')
time.sleep(1)

这个应该可以。它将 e 设置为“ ,while 循环检查它是否仍然存在”。如果 try 语句捕获错误,它将输出连接被拒绝,等待1秒钟,然后重新开始。它将继续运行,直到 try 中没有错误,然后将 e 设置为’,从而终止 while 循环。

When retrying due to error, you should always:

  • 实现重试限制,否则您可能会在无限循环中被阻塞
  • 实现一个延迟,否则您将会对资源造成太大的影响,比如您的 CPU 或者已经出现故障的远程服务器

解决这个问题的一个简单的通用方法是使用 退后库,同时涵盖这些问题。一个基本的例子:

import backoff


@backoff.on_exception(
backoff.expo,
MyException,
max_tries=5
)
def make_request(self, data):
# do the request

This code wraps make_request with a decorator which implements the retry logic. We retry whenever our specific error MyException occurs, with a limit of 5 retries. 截断二进制指数避退算法 is a good idea in this context to help minimize the additional burden our retries place on the remote server.

what about the 重试 Pypi 上的库? 我使用它已经有一段时间了,它可以做我想做的事情,甚至更多(在出错时重试,在没有时重试,在超时时重试)。下面是他们网站上的例子:

import random
from retrying import retry


@retry
def do_something_unreliable():
if random.randint(0, 10) > 1:
raise IOError("Broken sauce, everything is hosed!!!111one")
else:
return "Awesome sauce!"


print do_something_unreliable()

我现在正在尝试这个,这是我想出来的;

    placeholder = 1
while placeholder is not None:
try:
#Code
placeholder = None
except Exception as e:
print(str(datetime.time(datetime.now()))[:8] + str(e)) #To log the errors
placeholder = e
time.sleep(0.5)
continue