是否有一种pythonic方式尝试最多次？

Question

我有一个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子句中再次尝试来做到这一点,但这非常难看,我觉得必须有一个体面的方法来实现这一点.

Answer 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])

Answer 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

Answer 3

更新:有一个更好维护的重试库fork,称为tenacity,它支持更多功能,并且通常更灵活.

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

一些例子:

@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"

Answer 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])

Answer 5

我会像这样重构它:

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函数分解似乎打破了功能,以便很容易看到业务逻辑而不会陷入重试代码中.

Answer 6

像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