在 Python 中尝试直到没有错误

Question

我有一段 Python 代码似乎在概率上会导致错误，因为它正在访问服务器，有时该服务器有 500 内部服务器错误。我想继续尝试，直到我没有收到错误。我的解决方案是：

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

这对我来说似乎是一种黑客行为。有没有更 Pythonic 的方式来做到这一点？

Answer 1

它不会变得更干净。这不是一件很干净的事情。充其量（无论如何这将更具可读性，因为break 的条件与while 一致），您可以创建一个变量result = None 并在它is None 时循环。您还应该调整变量，您可以将continue 替换为语义上可能正确的pass（您不关心是否发生错误，您只想忽略它）并删除break - 这也可以获得其余的代码，只执行一次，在循环之外。另请注意，由于given in the documentation 的原因，裸露的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

Answer 2

这是一个在 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 str_error:
        pass

    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

Answer 3

可能是这样的：

connected = False

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

Answer 4

当因错误重试时，您应该始终：

• 实施重试限制，否则您可能会被无限循环阻塞
• 实施延迟，否则您将过度使用资源，例如您的 CPU 或已经陷入困境的远程服务器

在解决这些问题的同时解决此问题的一种简单通用方法是使用backoff 库。一个基本的例子：

import backoff

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

这段代码用实现重试逻辑的装饰器包装了 make_request。每当发生特定错误MyException 时，我们都会重试，重试次数限制为 5 次。在这种情况下，Exponential backoff 是一个好主意，可以帮助最大限度地减少重试对远程服务器造成的额外负担。

Answer 5

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)

此时，通过循环或调用 next() 将其视为任何其他迭代器。

Answer 6

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

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)

Answer 7

也许是基于装饰器的？ 您可以将我们想要重试的异常列表和/或尝试次数作为装饰器参数传递。

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()

当然，“try”部分应该移到另一个函数，因为我们在两个循环中都使用它，但这只是示例；）

Answer 8

与大多数其他人一样，我建议尝试有限次数并在尝试之间休眠。这样一来，您就不会陷入无限循环，以防远程服务器发生实际情况。

我还建议您仅在遇到预期的特定异常时才继续。这样，您仍然可以处理您可能没有预料到的异常。

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 块。由于在 except 块中继续，您跳过循环的其余部分，直到 try 块工作，满足 while 条件，或者出现 HTTPError 以外的异常。

Answer 9

retrying library on 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()

Answer 10

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 循环。

Answer 11

我现在正在尝试这个，这就是我想出的；

    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

Answer 12

这是我用来将错误捕获为字符串的一小段代码。将重试直到成功。这会捕获所有异常，但您可以根据需要进行更改。

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

警告：此代码将一直卡在一个循环中，直到没有异常发生。这只是一个简单的示例，可能需要您尽快跳出循环或在重试之间休眠。