¿Existe una forma pitónica de probar algo hasta un número máximo de veces? [duplicar]

Tengo un script de Python que consulta 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. Entonces, me gustaría saber si hay una forma sensata en Python de intentar ejecutar una consulta y, si falla, intentarlo nuevamente, hasta un número fijo de intentos. Probablemente me gustaría intentarlo 5 veces antes de rendirme por completo.

Este es 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 con otro intento en la cláusula excepto, 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 usa el decoratormódulo

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 decoratormódulo :

$ easy_install decorator

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

Sí, existe la biblioteca de reintentos , que tiene un decorador que implementa varios tipos de lógica de reintentos que puedes 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 calleefunción parece romper la funcionalidad de modo que sea fácil ver la lógica empresarial sin atascarse en el código de reintento.

Como 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.

Puede usar un forbucle con una elsecláusula para obtener el máximo efecto:

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

La clave es salir del ciclo tan pronto como la consulta tenga éxito. La elsecláusula solo se activará si el ciclo se completa sin un break.