有没有一种 Python 式的方法可以尝试最多次数？-IT科技

摘要：问题描述：我有一个 Python 脚本，它正在查询共享 Linux 主机上的 MySQL 服务器。由于某种原因，查询 MySQL 时经常返回“服务器已消失”错误：_mysql_exceptions.OperationalError: (2006, 'MySQL server has gone away') 如...

问题描述：

我有一个 Python 脚本，它正在查询共享 Linux 主机上的 MySQL 服务器。由于某种原因，查询 MySQL 时经常返回“服务器已消失”错误：

_mysql_exceptions.OperationalError: (2006, 'MySQL server has gone away')

如果您之后立即重试该查询，通常会成功。所以，我想知道在 Python 中是否有一种合理的方法可以尝试执行查询，如果失败，则重试，最多尝试固定次数。我可能希望它尝试 5 次后就完全放弃。

以下是我拥有的代码类型：

conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()

try:
    cursor.execute(query)
    rows = cursor.fetchall()
    for row in rows:
        # do something with the data
except MySQLdb.Error, e:
    print "MySQL Error %d: %s" % (e.args[0], e.args[1])

显然，我可以通过在 except 子句中进行另一次尝试来做到这一点，但这非常丑陋，而且我觉得一定有一种不错的方法来实现这一点。

解决方案 1：

怎么样：

conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
attempts = 0

while attempts < 3:
    try:
        cursor.execute(query)
        rows = cursor.fetchall()
        for row in rows:
            # do something with the data
        break
    except MySQLdb.Error, e:
        attempts += 1
        print "MySQL Error %d: %s" % (e.args[0], e.args[1])

解决方案 2：

根据 Dana 的回答，您可能希望以装饰者的身份来做这件事：

def retry(howmany):
    def tryIt(func):
        def f():
            attempts = 0
            while attempts < howmany:
                try:
                    return func()
                except:
                    attempts += 1
        return f
    return tryIt

然后...

@retry(5)
def the_db_func():
    # [...]

使用该`decorator`模块的增强版本

import decorator, time

def retry(howmany, *exception_types, **kwargs):
    timeout = kwargs.get('timeout', 0.0) # seconds
    @decorator.decorator
    def tryIt(func, *fargs, **fkwargs):
        for _ in xrange(howmany):
            try: return func(*fargs, **fkwargs)
            except exception_types or Exception:
                if timeout is not None: time.sleep(timeout)
    return tryIt

然后...

@retry(5, MySQLdb.Error, timeout=0.5)
def the_db_func():
    # [...]

要安装模块decorator：

$ easy_install decorator

解决方案 3：

更新：有一个维护得更好的重试库分支，称为tenacity，它支持更多功能并且通常更灵活。

API 略有变化：

@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
    print("Stopping after 7 attempts")

@retry(wait=wait_fixed(2))
def wait_2_s():
    print("Wait 2 second between retries")

@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1000():
    print("Wait 2^x * 1000 milliseconds between each retry,")
    print("up to 10 seconds, then 10 seconds afterwards")

是的，有一个重试库，它有一个装饰器，可以实现几种可以组合的重试逻辑：

一些例子：

@retry(stop_max_attempt_number=7)
def stop_after_7_attempts():
    print("Stopping after 7 attempts")

@retry(wait_fixed=2000)
def wait_2_s():
    print("Wait 2 second between retries")

@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)
def wait_exponential_1000():
    print("Wait 2^x * 1000 milliseconds between each retry,")
    print("up to 10 seconds, then 10 seconds afterwards")

解决方案 4：

conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()

for i in range(3):
    try:
        cursor.execute(query)
        rows = cursor.fetchall()
        for row in rows:
            # do something with the data
        break
    except MySQLdb.Error, e:
        print "MySQL Error %d: %s" % (e.args[0], e.args[1])

解决方案 5：

像 S.Lott 一样，我喜欢用标志来检查我们是否完成了：

conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()

success = False
attempts = 0

while attempts < 3 and not success:
    try:
        cursor.execute(query)
        rows = cursor.fetchall()
        for row in rows:
            # do something with the data
        success = True 
    except MySQLdb.Error, e:
        print "MySQL Error %d: %s" % (e.args[0], e.args[1])
        attempts += 1

解决方案 6：

我会像这样重构它：

def callee(cursor):
    cursor.execute(query)
    rows = cursor.fetchall()
    for row in rows:
        # do something with the data

def caller(attempt_count=3, wait_interval=20):
    """:param wait_interval: In seconds."""
    conn = MySQLdb.connect(host, user, password, database)
    cursor = conn.cursor()
    for attempt_number in range(attempt_count):
        try:
            callee(cursor)
        except MySQLdb.Error, e:
            logging.warn("MySQL Error %d: %s", e.args[0], e.args[1])
            time.sleep(wait_interval)
        else:
            break

分解callee函数似乎会分解功能，以便轻松查看业务逻辑，而不会陷入重试代码中。

解决方案 7：

您可以使用for带有else子句的循环来获得最佳效果：

conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()

for n in range(3):
    try:
        cursor.execute(query)
    except MySQLdb.Error, e:
        print "MySQL Error %d: %s" % (e.args[0], e.args[1])
    else:
        rows = cursor.fetchall()
        for row in rows:
            # do something with the data
        break
else:
    # All attempts failed, raise a real error or whatever

关键在于查询成功后立即跳出循环。else仅当循环完成且没有时，才会触发该子句break。

解决方案 8：

def successful_transaction(transaction):
    try:
        transaction()
        return True
    except SQL...:
        return False

succeeded = any(successful_transaction(transaction)
                for transaction in repeat(transaction, 3))

解决方案 9：

1.定义：

def try_three_times(express):
    att = 0
    while att < 3:
        try: return express()
        except: att += 1
    else: return u"FAILED"

2.使用方法：

try_three_times(lambda: do_some_function_or_express())

我用它来解析 html 上下文。

解决方案 10：

有一些很棒的 Python 包专门用于重试逻辑：

耐力
韧性
退避

耐力示例

import stamina

@stamina.retry(on=(MyPossibleException1, 
                   MyPossibleException2), 
               attempts=3)
def your_function(param1, param2):
    # Do something

坚韧不拔的典范

from tenacity import wait_exponential, retry, stop_after_attempt

@retry(wait=wait_exponential(multiplier=2, min=2, max=30),
       stop=stop_after_attempt(5))
def your_function(param1, param2):
   # Do something

退避示例

import backoff

@backoff.on_exception(backoff.expo,
                      (MyPossibleException1,
                       MyPossibleException2))
def your_function(param1, param2):
    # Do something

解决方案 11：

这是我的通用解决方案：

class TryTimes(object):
    ''' A context-managed coroutine that returns True until a number of tries have been reached. '''

    def __init__(self, times):
        ''' times: Number of retries before failing. '''
        self.times = times
        self.count = 0

    def __next__(self):
        ''' A generator expression that counts up to times. '''
        while self.count < self.times:
            self.count += 1
        yield False

    def __call__(self, *args, **kwargs):
        ''' This allows "o() calls for "o = TryTimes(3)". '''
        return self.__next__().next()

    def __enter__(self):
        ''' Context manager entry, bound to t in "with TryTimes(3) as t" '''
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        ''' Context manager exit. '''
        return False # don't suppress exception

这允许如下代码：

with TryTimes(3) as t:
    while t():
        print "Your code to try several times"

也可能：

t = TryTimes(3)
while t():
    print "Your code to try several times"

我希望能够通过更直观的方式处理异常来改进这一点。欢迎提出建议。