¿Existe una forma pythonica de probar algo hasta un número máximo de veces?

Tengo un script en Python que está consultando un servidor MySQL en un host linux compartido. Por alguna razón, las consultas a MySQL a menudo devuelven un error de “el servidor se ha ido”:

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

Si vuelve a intentar la consulta inmediatamente después, normalmente se realiza correctamente. Por lo tanto, me gustaría saber si hay una forma razonable en Python para intentar ejecutar una consulta y, si falla, volver a intentarlo, hasta un número fijo de bashs. Probablemente me gustaría intentarlo 5 veces antes de rendirme por completo.

Aquí está el tipo de código que tengo:

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

Claramente, podría hacerlo haciendo otro bash en la cláusula de excepción, pero eso es increíblemente feo, y tengo la sensación de que debe haber una manera decente de lograrlo.

Qué tal si:

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

Sobre la base de la respuesta de Dana, es posible que desee hacer esto como decorador:

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

Entonces...

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

Versión mejorada que utiliza el módulo 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 

Entonces...

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

Para instalar el módulo decorator :

 $ easy_install decorator 

ACTUALIZACIÓN: hay una bifurcación mejor mantenida de la biblioteca de rebashs llamada tenacidad , que admite más funciones y, en general, es más flexible.


Sí, existe la biblioteca de rebashs , que tiene un decorador que implementa varios tipos de lógica de rebashs que puede combinar:

Algunos ejemplos:

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

Lo refactorizaría así:

 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 

Factorizar la función de la persona callee parece romper la funcionalidad de modo que es fácil ver la lógica de negocios sin atascarse en el código de rebash.

Al igual que S.Lott, me gusta una bandera para comprobar si hemos terminado:

 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 successful_transaction(transaction): try: transaction() return True except SQL...: return False succeeded = any(successful_transaction(transaction) for transaction in repeat(transaction, 3)) 

1.Definición:

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

2. Uso:

 try_three_times(lambda: do_some_function_or_express()) 

Lo uso para analizar el contexto html.

Esta es mi solución genérica:

 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 

Esto permite código como el siguiente:

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

También es posible:

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

Esto se puede mejorar manejando las excepciones de una manera más intuitiva, espero. Abierto a sugerencias.