tenacity
tenacity 9.0.0
Consulte la documentación de tenacidad para una mejor experiencia.
Tenacity es una biblioteca de reintentos de uso general con licencia Apache 2.0, escrita en Python, para simplificar la tarea de agregar comportamiento de reintento a casi cualquier cosa. Se origina en una bifurcación de reintento que lamentablemente ya no se mantiene. Tenacity no es compatible con la API con el reintento, pero agrega nuevas funciones importantes y corrige una serie de errores de larga data.
El caso de uso más simple es volver a intentar una función inestable cada vez que ocurre una excepción hasta que se devuelve un valor.
.. código de prueba::
importar aleatoriamente
desde el reintento de importación de tenacidad
@rever
def hacer_algo_no confiable():
si aleatorio.randint(0, 10) > 1:
elevar IOError ("Salsa rota, ¡¡¡todo está regado!!! 111uno")
demás:
volver "¡Salsa increíble!"
imprimir(hacer_algo_no confiable())
.. salida de prueba:: :esconder: Salsa impresionante!
.. toctree::
:oculto:
: profundidad máxima: 2
registro de cambios
API
Para instalar tenacity , simplemente:
$ pip install tenacity .. configuración de prueba::
importar registro
#
# Tenga en cuenta que la siguiente importación se utiliza únicamente para fines de demostración.
# El código de producción siempre debe importar explícitamente los nombres que necesita.
#
de importación de tenacidad *
clase MiExcepción(Excepción):
aprobar
Como vio anteriormente, el comportamiento predeterminado es volver a intentarlo siempre sin esperar cuando se genera una excepción.
.. código de prueba::
@rever
def nunca_gonna_give_you_up():
print("Reintentar siempre ignorando las excepciones, no esperar entre reintentos")
levantar excepción
Seamos un poco menos persistentes y establezcamos algunos límites, como el número de intentos antes de rendirnos.
.. código de prueba::
@reintentar(parar=stop_after_attempt(7))
def stop_after_7_attempts():
print("Parando después de 7 intentos")
levantar excepción
No tenemos todo el día, así que establezcamos un límite de cuánto tiempo deberíamos volver a intentar cosas.
.. código de prueba::
@reintentar(parar=parar_después_delay(10))
definición stop_after_10_s():
print("Parando después de 10 segundos")
levantar excepción
Si tiene una fecha límite ajustada y exceder el tiempo de demora no está bien, entonces puede renunciar a los reintentos un intento antes de exceder la demora.
.. código de prueba::
@reintentar(parar=parar_antes_delay(10))
definición stop_before_10_s():
print("Deteniendo 1 intento antes de 10 segundos")
levantar excepción
Puede combinar varias condiciones de parada utilizando el | operador:
.. código de prueba::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("Parando después de 10 segundos o 5 reintentos")
levantar excepción
A la mayoría de las cosas no les gusta que las sondeen lo más rápido posible, así que esperemos 2 segundos entre reintentos.
.. código de prueba::
@reintentar(esperar=esperar_fixed(2))
def esperar_2_s():
print("Esperar 2 segundos entre reintentos")
levantar excepción
Algunas cosas funcionan mejor con un poco de aleatoriedad inyectada.
.. código de prueba::
@retry(esperar=esperar_aleatorio(mín=1, máximo=2))
def esperar_aleatorio_1_to_2_s():
print("Esperar aleatoriamente de 1 a 2 segundos entre reintentos")
levantar excepción
Por otra parte, es difícil superar el retroceso exponencial al volver a intentar servicios distribuidos y otros puntos finales remotos.
.. código de prueba::
@retry(wait=wait_exponential(multiplicador=1, min=4, max=10))
def esperar_exponencial_1():
print("Espere 2^x * 1 segundo entre cada reintento, comenzando con 4 segundos, luego hasta 10 segundos, luego 10 segundos después")
levantar excepción
Por otra parte, también es difícil superar la combinación de esperas fijas y jitter (para ayudar a evitar rebaños atronadores) al reintentar servicios distribuidos y otros puntos finales remotos.
.. código de prueba::
@reintentar(esperar=esperar_fijo(3) + esperar_aleatorio(0, 2))
def esperar_fija_jitter():
print("Espere al menos 3 segundos y agregue hasta 2 segundos de retraso aleatorio")
levantar excepción
Cuando varios procesos compiten por un recurso compartido, el aumento exponencial de la fluctuación ayuda a minimizar las colisiones.
.. código de prueba::
@retry(wait=wait_random_exponential(multiplicador=1, max=60))
def esperar_jitter_exponencial():
print("Espere aleatoriamente hasta 2^x * 1 segundos entre cada reintento hasta que el rango alcance los 60 segundos, luego aleatoriamente hasta 60 segundos después")
levantar excepción
A veces es necesario construir una cadena de retrocesos.
.. código de prueba::
@retry(wait=wait_chain(*[wait_fixed(3) para i en el rango(3)] +
[wait_fixed(7) para i en el rango(2)] +
[wait_fixed(9)]))
def esperar_fixed_chained():
print("Espera 3 segundos para 3 intentos, 7 segundos para los siguientes 2 intentos y 9 segundos para todos los intentos posteriores")
levantar excepción
Tenemos algunas opciones para lidiar con reintentos que generan excepciones específicas o generales, como en los casos aquí.
.. código de prueba::
clase ClientError (excepción):
"""Algún tipo de error del cliente."""
@reintentar(reintentar=reintentar_if_excepción_tipo(IOError))
def podría_io_error():
print("Vuelva a intentarlo siempre sin esperar si se produce un error IOE, genere cualquier otro error")
levantar excepción
@reintentar(reintentar=reintentar_if_not_exception_type(ClientError))
def might_client_error():
print("Vuelva a intentarlo siempre sin esperar si ocurre algún error que no sea ClientError. Inicie inmediatamente ClientError.")
levantar excepción
También podemos usar el resultado de la función para alterar el comportamiento del reintento.
.. código de prueba::
def es_none_p(valor):
"""Devuelve Verdadero si el valor es Ninguno"""
el valor de retorno es Ninguno
@reintentar(reintentar=reintentar_if_result(is_none_p))
def might_return_none():
print("Reintentar sin espera si el valor de retorno es Ninguno")
Vea también estos métodos:
.. código de prueba::
reintento_si_excepción
reintentar_si_tipo_excepción
reintentar_si_no_tipo_excepción
reintento_a menos que_excepción_tipo
reintentar_si_resultado
reintentar_si_no_resultado
reintentar_si_excepción_mensaje
reintentar_si_no_excepción_mensaje
reintentar_cualquiera
reintentar_todos
También podemos combinar varias condiciones:
.. código de prueba::
def es_none_p(valor):
"""Devuelve Verdadero si el valor es Ninguno"""
el valor de retorno es Ninguno
@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
print("Reintentar siempre ignorando las excepciones sin esperar si el valor de retorno es Ninguno")
También se admite cualquier combinación de parada, espera, etc. para darle la libertad de mezclar y combinar.
También es posible volver a intentarlo explícitamente en cualquier momento generando la excepción TryAgain:
.. código de prueba::
@rever
def hacer_algo():
resultado = algo_más()
si resultado == 23:
aumentar Inténtalo de nuevo
Normalmente, cuando su función falla por última vez (y no se volverá a intentar según su configuración), se genera un RetryError. La excepción que encontró su código se mostrará en algún lugar en el medio del seguimiento de la pila.
Si prefiere ver la excepción que encontró su código al final del seguimiento de la pila (donde es más visible), puede establecer reraise=True.
.. código de prueba::
@retry(reaise=Verdadero, stop=stop_after_attempt(3))
def elevar_mi_excepción():
generar MyException ("Fallo")
intentar:
elevar_mi_excepción()
excepto MiExcepción:
# se agotó el tiempo de reintento
aprobar
Es posible ejecutar una acción antes de cualquier intento de llamar a la función usando la función de devolución de llamada anterior:
.. código de prueba::
importar registro
sistema de importación
logging.basicConfig(flujo=sys.stderr, nivel=logging.DEBUG)
registrador = logging.getLogger(__nombre__)
@retry(stop=stop_after_attempt(3), before=before_log(registrador, logging.DEBUG))
def elevar_mi_excepción():
generar MyException ("Fallo")
Del mismo modo, es posible ejecutar después de una llamada fallida:
.. código de prueba::
importar registro
sistema de importación
logging.basicConfig(flujo=sys.stderr, nivel=logging.DEBUG)
registrador = logging.getLogger(__nombre__)
@retry(stop=stop_after_attempt(3), after=after_log(registrador, logging.DEBUG))
def elevar_mi_excepción():
generar MyException ("Fallo")
También es posible registrar únicamente los errores que se van a reintentar. Normalmente los reintentos ocurren después de un intervalo de espera, por lo que el argumento de palabra clave se llama before_sleep :
.. código de prueba::
importar registro
sistema de importación
logging.basicConfig(flujo=sys.stderr, nivel=logging.DEBUG)
registrador = logging.getLogger(__nombre__)
@retry(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(registrador, logging.DEBUG))
def elevar_mi_excepción():
generar MyException ("Fallo")
Puede acceder a las estadísticas sobre el reintento realizado a través de una función utilizando el atributo de estadísticas adjunto a la función:
.. código de prueba::
@reintentar(parar=stop_after_attempt(3))
def elevar_mi_excepción():
generar MyException ("Fallo")
intentar:
elevar_mi_excepción()
excepto Excepción:
aprobar
imprimir(raise_my_exception.estadísticas)
.. salida de prueba:: :esconder: ...
También puede definir sus propias devoluciones de llamada. La devolución de llamada debe aceptar un parámetro llamado retry_state que contiene toda la información sobre la invocación de reintento actual.
Por ejemplo, puede llamar a una función de devolución de llamada personalizada después de que todos los reintentos fallaron, sin generar una excepción (o puede volver a generarla o hacer algo realmente).
.. código de prueba::
def retorno_último_valor (reintento_estado):
"""devuelve el resultado del último intento de llamada"""
devolver retry_state.outcome.result()
def es_falso(valor):
"""Devuelve Verdadero si el valor es Falso"""
el valor de retorno es falso
# devolverá Falso después de intentar 3 veces obtener un resultado diferente
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
reintentar=reintentar_if_resultado(es_falso))
def eventualmente_return_false():
devolver falso
El argumento retry_state es un objeto de la clase :class:`~tenacity.RetryCallState` .
También es posible definir devoluciones de llamada personalizadas para otros argumentos de palabras clave.
.. función:: my_stop(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual :return: si el reintento debe detenerse o no :rtipo: booleano
.. función:: my_wait(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual :return: número de segundos a esperar antes del próximo reintento :rtipo: flotante
.. función:: my_retry(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual :return: si el reintento debe continuar o no :rtipo: booleano
.. función:: my_before(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual
.. función:: my_after(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual
.. función:: my_before_sleep(retry_state) :param RetryCallState retry_state: información sobre la invocación de reintento actual
Aquí hay un ejemplo con una función personalizada before_sleep :
.. código de prueba::
importar registro
logging.basicConfig(flujo=sys.stderr, nivel=logging.DEBUG)
registrador = logging.getLogger(__nombre__)
def my_before_sleep(retry_state):
si estado_reintento.número_intento < 1:
nivel de registro = registro.INFO
demás:
nivel de registro = registro.ADVERTENCIA
registrador.log(
loglevel, 'Reintentando %s: el intento %s terminó con: %s',
reintento_estado.fn, reintento_estado.intento_número, reintento_estado.resultado)
@retry(stop=stop_after_attempt(3), before_sleep=mi_antes_de_dormir)
def elevar_mi_excepción():
generar MyException ("Fallo")
intentar:
elevar_mi_excepción()
excepto ReintentarError:
aprobar
Puede cambiar los argumentos de un decorador de reintento según sea necesario al llamarlo utilizando la función retry_with adjunta a la función envuelta:
.. código de prueba::
@reintentar(parar=stop_after_attempt(3))
def elevar_mi_excepción():
generar MyException ("Fallo")
intentar:
elevar_mi_excepción.retry_with(stop=stop_after_attempt(4))()
excepto Excepción:
aprobar
imprimir(raise_my_exception.estadísticas)
.. salida de prueba:: :esconder: ...
Si desea utilizar variables para configurar los parámetros de reintento, no es necesario utilizar el decorador de reintento; en su lugar, puede utilizar Reintentar directamente:
.. código de prueba::
def nunca_bueno_enough(arg1):
generar excepción ('argumento no válido: {}'.format(arg1))
def try_never_good_enough(max_attempts=3):
reintento = Reintentando(stop=stop_after_attempt(max_attempts), rereaise=True)
reintentar(never_good_enough, 'Realmente lo intento')
Es posible que también desees cambiar temporalmente el comportamiento de una función decorada, como en las pruebas, para evitar tiempos de espera innecesarios. Puede modificar/parchear el atributo de reintento adjunto a la función. Tenga en cuenta que este es un atributo de solo escritura; las estadísticas deben leerse desde el atributo de estadísticas de la función.
.. código de prueba::
@retry(stop=stop_after_attempt(3), espera=wait_fixed(3))
def elevar_mi_excepción():
generar MyException ("Fallo")
del simulacro de importación unittest
con simulado.patch.object(raise_my_exception.retry, "esperar", esperar_fixed(0)):
intentar:
elevar_mi_excepción()
excepto Excepción:
aprobar
imprimir(raise_my_exception.estadísticas)
.. salida de prueba:: :esconder: ...
Tenacity le permite volver a intentar un bloque de código sin la necesidad de envolverlo en una función aislada. Esto facilita aislar el bloque defectuoso mientras se comparte el contexto. El truco consiste en combinar un bucle for y un administrador de contexto.
.. código de prueba::
de tenacity import Reintentar, RetryError, stop_after_attempt
intentar:
para intentar reintentar (stop=stop_after_attempt(3)):
con intento:
generar excepción ('¡Mi código está fallando!')
excepto ReintentarError:
aprobar
Puede configurar todos los detalles de la política de reintento configurando el objeto Reintento.
Con código asíncrono puedes usar AsyncRetrying.
.. código de prueba::
desde tenacidad importa AsyncRetrying, RetryError, stop_after_attempt
función de definición asíncrona():
intentar:
async para intento en AsyncRetrying(stop=stop_after_attempt(3)):
con intento:
generar excepción ('¡Mi código está fallando!')
excepto ReintentarError:
aprobar
En ambos casos, es posible que desees establecer el resultado del intento para que esté disponible en estrategias de reintento como retry_if_result . Esto se puede hacer accediendo a la propiedad retry_state :
.. código de prueba::
desde tenacidad importar AsyncRetrying, retry_if_result
función de definición asíncrona():
async para intento en AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
con intento:
resultado = 1 # Algún cálculo complejo, llamada a función, etc.
si no, intente.retry_state.outcome.failed:
intento.retry_state.set_result(resultado)
resultado de retorno
Finalmente, retry también funciona en las corrutinas asyncio, Trio y Tornado (>= 4.5). Los sueños también se realizan de forma asincrónica.
@ retry
async def my_asyncio_function ( loop ):
await loop . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
async def my_async_trio_function ():
await trio . socket . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
@ tornado . gen . coroutine
def my_async_tornado_function ( http_client , url ):
yield http_client . fetch ( url )Incluso puedes usar bucles de eventos alternativos como curio pasando la función de suspensión correcta:
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno se utiliza para gestionar registros de cambios. Eche un vistazo a sus documentos de uso.
La generación de documentos compilará automáticamente los registros de cambios. Sólo necesitas agregarlos.
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
