Conquistar reintentos en Python utilizando Tenacity Un tutorial de principio a fin

'Python Retry Conquests with Tenacity A Comprehensive Tutorial'

CAJA DE HERRAMIENTAS DE PYTHON

Mejorando tus proyectos de Python con mecanismos de reintentos robustos y técnicas de manejo de errores

Este artículo discutirá el uso básico y las capacidades de personalización de Tenacity.

Esta biblioteca de Python instrumental proporciona un mecanismo de reintentos. También exploraremos las capacidades de reintentos y manejo de excepciones de Tenacity a través de un ejemplo práctico.

Introducción

Imagina que estás administrando cientos de servicios web, algunos ubicados en el extranjero (con alta latencia) y otros bastante antiguos (y no muy estables). ¿Cómo te sentirías?

Mi colega Wang está en esa situación. Me dijo que estaba bastante frustrado:

• Cómo construir una plataforma de análisis semi-estructurado en tiempo real en Snowflake
• ¿Es la Ciencia de Datos una buena carrera?
• ¿Cómo cambiar de carrera de analista de datos a científico de datos?

Todos los días, necesita verificar el estado de invocación de estos servicios remotos, y a menudo se encuentra con problemas de tiempo de espera u otras anomalías. La solución de problemas es particularmente desafiante.

Además, gran parte del código del lado del cliente fue escrito por sus predecesores, lo que dificulta realizar refactorizaciones sustanciales. Por lo tanto, los servicios deben seguir funcionando tal como están.

Sería genial si hubiera una forma de reconectar automáticamente estas llamadas remotas después de que ocurra una excepción. Con lágrimas en los ojos, Wang me miró.

Le aseguré que no había problema y le presenté una nueva herramienta de mi caja de herramientas: Tenacity. Con solo un decorador, el código existente puede obtener capacidades de reintentos. Veamos cómo usarlo.

Instalación y Uso Básico

Dado que el sitio web oficial de Tenacity solo ofrece un documento API simple, comencemos con la instalación de la biblioteca y algunos usos básicos.

Instalación

Si estás usando pip, simplemente ejecuta lo siguiente:

python -m pip install tenacity

Si estás usando Anaconda, Tenacity no está en el canal predeterminado, por lo que debes instalarlo desde conda-forge:

conda install -c conda-forge tenacity

Uso básico

Después de instalar Tenacity, veamos algunos usos básicos de la biblioteca.

Simplemente agrega un decorador @retry y tu código tendrá capacidades de reintentos:

@retry()async def coro_func():    pass

Si deseas que tu código deje de reintentar después de cierto número de intentos, puedes escribirlo de la siguiente manera:

@retry(stop=stop_after_attempt(5))async def coro_func():    pass

Por supuesto, para evitar reintentos frecuentes que puedan agotar los grupos de conexiones, recomiendo agregar un tiempo de espera antes de cada reintento. Por ejemplo, si deseas esperar 2 segundos antes de cada conexión:

@retry(wait=wait_fixed(2))async def coro_func():    pass

Aunque no se menciona en la documentación, prefiero esperar un segundo extra más que la última vez antes de cada reintento para minimizar el desperdicio de recursos:

@retry(wait=wait_incrementing(start=1, increment=1, max=5))async def coro_func():    pass

Finalmente, si el reintento es causado por una excepción que se lanza en el método, es mejor lanzar la excepción de nuevo. Esto permite un manejo de excepciones más flexible al llamar al método:

@retry(reraise=True, stop=stop_after_attempt(3))async def coro_func():    pass

Funciones Avanzadas: Callbacks Personalizados

Además de algunos casos de uso comunes, puedes agregar tu propia lógica de determinación de reintento, como decidir en función del resultado de la ejecución del método o imprimir los parámetros de invocación del método antes de la ejecución.

En este caso, podemos usar Callbacks Personalizados para personalización.

Existen dos formas de extender Custom Callbacks:

Una es el enfoque recomendado en la documentación: escribir un método de extensión.

Este método recibirá una instancia de RetryCallState como parámetro cuando se ejecute.

A través de este parámetro, podemos obtener el método encapsulado, los parámetros de la llamada al método, el resultado devuelto y cualquier excepción lanzada.

Por ejemplo, podemos usar este enfoque para juzgar el valor de retorno de un método y reintentar si el valor es par:

from tenacity import *
def check_is_even(retry_state: RetryCallState):
    if retry_state.outcome.exception():
        return True
    return retry_state.outcome.result() % 2 == 0

Por supuesto, antes de hacer este juicio, si se lanza una excepción, reintentamos directamente.

Si necesitas pasar parámetros adicionales en el método de extensión, puedes agregar un envoltorio fuera del método de extensión.

Por ejemplo, este envoltorio pasará un parámetro logger. Cuando el número de reintentos supere dos, imprimirá el tiempo de reintentos, el nombre del método y los parámetros del método en el registro:

def my_before_log(logger: Logger):
    def my_log(retry_state: RetryCallState):
        fn = retry_state.fn
        args = retry_state.args
        attempt = retry_state.attempt_number
        if attempt > 2:
            logger.warning(f"Iniciando reintentos del método {fn.__name__} con args: {args}")
    return my_log

Ejemplo de Red del Mundo Real

Finalmente, para darte una profunda impresión de cómo usar Tenacity en tus proyectos, utilizaré un proyecto de cliente remoto como ejemplo para demostrar cómo integrar las potentes capacidades de Tenacity.

Este proyecto simulará el acceso a un servicio HTTP y decidirá si reintentar o no en función del código de estado devuelto.

Por supuesto, para evitar el desperdicio de recursos del servidor debido a tiempos de espera de conexión prolongados, también agregaré un tiempo de espera de 2 segundos para cada solicitud. Si ocurre un tiempo de espera, se volverá a intentar la conexión.

Antes de comenzar el código, implementaré varios métodos de extensión. Uno de los métodos consiste en juzgar cuándo el número de reintentos de un método supera dos y mostrar un mensaje de advertencia en el registro:

Otro método de extensión es juzgar el código de estado devuelto. Si el código de estado es mayor que 300, reintentar. Por supuesto, los tiempos de espera también desencadenarán reintentos.

A continuación, tenemos la implementación del método de llamada remota. Después de escribir el método, recuerda agregar el decorador de reintentos de Tenacity. La estrategia que utilizo aquí es reintentar hasta 20 veces, esperando 1 segundo más que el reintento anterior antes de cada reintento.

Por supuesto, no olvides agregar los dos métodos de extensión que acabamos de implementar:

¡Misión cumplida! ¿No fue súper sencillo?

Conclusión

Me alegra haber ayudado a Wang a resolver otro problema.

Al usar Tenacity, podemos equipar fácilmente código existente con varios mecanismos de reintentos, mejorando así la robustez y las capacidades de autorecuperación del programa.

Me haría aún más feliz si esta biblioteca pudiera ayudarte a resolver problemas. No dudes en dejar un comentario y discutir.

Además de mejorar la velocidad de ejecución y el rendimiento del código, el uso de varias herramientas para mejorar la eficiencia laboral también es un aumento de rendimiento:

Peng Qian

Caja de herramientas de Python

Ver lista de 4 historias

Únete a Zepes con mi enlace de referencia – Peng Qian

Como miembro de Zepes, una parte de tu tarifa de membresía se destina a los escritores que lees, y obtienes acceso completo a cada historia…

Zepes.com

We will continue to update Zepes; if you have any questions or suggestions, please contact us!