有没有一种pythonic的方式来尝试一些最多的次数？
我有一个python脚本查询共享linux主机上的MySQL服务器。 出于某种原因，对MySQL的查询经常返回一个“服务器已经消失”的错误：
_mysql_exceptions.OperationalError: (2006, 'MySQL server has gone away') 如果以后再次尝试查询，通常会成功。 所以，我想知道是否有一个明智的方式在python试图执行查询，如果失败，再试一次，达到固定数量的尝试。 大概我想要它在完全放弃之前尝试5次。
这是我有的代码types：
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])
显然我可以通过在除外条款中再次尝试，但这是非常难看的，我有一种感觉，必须有一个体面的方式来实现这一点。
怎么样：
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])
build立在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
更新：有一个更好的维护重试库称为韧性 ，它支持更多的function，一般更灵活。
是的，有重试库 ，它有一个装饰器，可以实现几种重试逻辑，你可以结合使用：
一些例子：
@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"
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])
像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
我会像这样重构它：
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函数似乎打破了function，因此很容易看到业务逻辑，而不会陷入重试代码中。
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())
我用它来parsinghtml上下文。
这是我的通用解决scheme：
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"
我希望这可以通过更直观的方式处理exception来改善。 打开build议。
def successful_transaction(transaction): try: transaction() return True except SQL...: return False succeeded = any(successful_transaction(transaction) for transaction in repeat(transaction, 3))
