Réessayez votre code Python jusqu’à ce qu’il échoue

Parfois, une fonction est appelée avec de mauvaises entrées ou dans un mauvais état de programme, elle échoue donc. Dans des langages comme Python, cela se traduit généralement par une exception.

Mais parfois, les exceptions sont causées par des problèmes différents ou sont transitoires. Imaginez un code qui doit continuer à fonctionner malgré le nettoyage des données de mise en cache. En théorie, le code et le nettoyeur pourraient soigneusement se mettre d’accord sur la méthodologie de nettoyage pour empêcher le code d’essayer d’accéder à un fichier ou un répertoire inexistant. Malheureusement, cette approche est compliquée et sujette aux erreurs. Cependant, la plupart de ces problèmes sont transitoires, car le nettoyeur finira par créer les structures correctes.

Encore plus fréquemment, la nature incertaine de la programmation réseau signifie que certaines fonctions qui résument un appel réseau échouent parce que des paquets ont été perdus ou corrompus.

Une solution courante consiste à réessayer le code défaillant. Cette pratique permet d’ignorer les problèmes de transition passés tout en échouant (éventuellement) si le problème persiste. Python dispose de plusieurs bibliothèques pour faciliter les nouvelles tentatives. Il s’agit d’un “exercice des doigts” courant.

Ténacité

Une bibliothèque qui va au-delà d’un exercice de doigté et dans une abstraction utile est ténacité. Installez-le avec pip install tenacity ou en dépendre en utilisant un dependencies = tenacity ligne dans votre pyproject.toml déposer.

Configurer la journalisation

Une fonction intégrée pratique de tenacity est un support pour la journalisation. Avec la gestion des erreurs, voir les détails du journal sur les nouvelles tentatives est inestimable.

Pour permettre aux exemples restants d’afficher des messages de journal, configurez la bibliothèque de journalisation. Dans un programme réel, le point d’entrée central ou un plugin de configuration de journalisation le fait. Voici un échantillon :

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s:%(name)s:%(levelname)s:%(message)s",
)

TENACITY_LOGGER = logging.getLogger("Retrying")

Échec sélectif

Pour montrer les fonctionnalités de tenacity, il est utile d’avoir un moyen d’échouer plusieurs fois avant de finalement réussir. En utilisant unittest.mock est utile pour ce scénario.

from unittest import mock

thing = mock.MagicMock(side_effect=[ValueError(), ValueError(), 3])

Si vous débutez dans les tests unitaires, lisez mon article sur mock.

Avant de montrer la puissance de tenacity, regardez ce qui se passe lorsque vous implémentez une nouvelle tentative directement dans une fonction. Démontrer cela permet de voir facilement l’effort manuel en utilisant tenacity sauve.

def useit(a_thing):
    for i in range(3):
        try:
            value = a_thing()
        except ValueError:
            TENACITY_LOGGER.info("Recovering")
            continue
        else:
            break
    else:
        raise ValueError()
    print("the value is", value)

La fonction peut être appelée avec quelque chose qui n’échoue jamais :

>>> useit(lambda: 5)
the value is 5

Avec la chose finalement réussie:

>>> useit(thing)

2023-03-29 17:00:42,774:Retrying:INFO:Recovering
2023-03-29 17:00:42,779:Retrying:INFO:Recovering

the value is 3

Appeler la fonction avec quelque chose qui échoue trop souvent se termine mal :

try:
    useit(mock.MagicMock(side_effect=[ValueError()] * 5 + [4]))
except Exception as exc:
    print("could not use it", repr(exc))

Le résultat:

2023-03-29 17:00:46,763:Retrying:INFO:Recovering
2023-03-29 17:00:46,767:Retrying:INFO:Recovering
2023-03-29 17:00:46,770:Retrying:INFO:Recovering

could not use it ValueError()

Utilisation simple de la ténacité

Pour la plupart, la fonction ci-dessus réessayait le code. L’étape suivante consiste à demander à un décorateur de gérer la logique de nouvelle tentative :

import tenacity

my_retry=tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
)

Tenacity prend en charge un nombre spécifié de tentatives et de journalisation après avoir obtenu une exception.

Le useit la fonction n’a plus à se soucier de réessayer. Parfois, il est logique que la fonction considère toujours réessayabilité. La ténacité permet au code de déterminer lui-même la possibilité de réessayer en levant l’exception spéciale TryAgain:

@my_retry
def useit(a_thing):
    try:
        value = a_thing()
    except ValueError:
        raise tenacity.TryAgain()
    print("the value is", value)

Maintenant, lors de l’appel useitil réessaie ValueError sans avoir besoin de code de nouvelle tentative personnalisé :

useit(mock.MagicMock(side_effect=[ValueError(), ValueError(), 2]))

Le résultat:

2023-03-29 17:12:19,074:Retrying:WARNING:Finished call to '__main__.useit' after 0.000(s), this was the 1st time calling it.
2023-03-29 17:12:19,080:Retrying:WARNING:Finished call to '__main__.useit' after 0.006(s), this was the 2nd time calling it.

the value is 2

Configurer le décorateur

Le décorateur ci-dessus n’est qu’un petit échantillon de ce que tenacity les soutiens. Voici un décorateur plus compliqué :

my_retry = tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
    before=tenacity.before_log(TENACITY_LOGGER, logging.WARNING),
    retry=tenacity.retry_if_exception_type(ValueError),
    wait=tenacity.wait_incrementing(1, 10, 2),
    reraise=True
)

Voici un exemple de décorateur plus réaliste avec des paramètres supplémentaires :

• before: Enregistrer avant d’appeler la fonction
• retry: Au lieu de simplement réessayer TryAgainréessayez les exceptions avec les critères donnés
• wait: Attendez entre les appels (ceci est particulièrement important si vous appelez un service)
• reraise: Si la nouvelle tentative a échoué, relance l’exception de la dernière tentative

Maintenant que le décorateur spécifie également la possibilité de réessayer, supprimez le code de useit:

@my_retry
def useit(a_thing):
    value = a_thing()
    print("the value is", value)

Voici comment cela fonctionne:

useit(mock.MagicMock(side_effect=[ValueError(), 5]))

Le résultat:

2023-03-29 17:19:39,820:Retrying:WARNING:Starting call to '__main__.useit', this is the 1st time calling it.
2023-03-29 17:19:39,823:Retrying:WARNING:Finished call to '__main__.useit' after 0.003(s), this was the 1st time calling it.
2023-03-29 17:19:40,829:Retrying:WARNING:Starting call to '__main__.useit', this is the 2nd time calling it.


the value is 5

Remarquez le délai entre les deuxième et troisième lignes du journal. C’est presque exactement une seconde:

>>> useit(mock.MagicMock(side_effect=[5]))

2023-03-29 17:20:25,172:Retrying:WARNING:Starting call to '__main__.useit', this is the 1st time calling it.

the value is 5

Avec plus de détails :

try:
    useit(mock.MagicMock(side_effect=[ValueError("detailed reason")]*3))
except Exception as exc:
    print("retrying failed", repr(exc))

Le résultat:

2023-03-29 17:21:22,884:Retrying:WARNING:Starting call to '__main__.useit', this is the 1st time calling it.
2023-03-29 17:21:22,888:Retrying:WARNING:Finished call to '__main__.useit' after 0.004(s), this was the 1st time calling it.
2023-03-29 17:21:23,892:Retrying:WARNING:Starting call to '__main__.useit', this is the 2nd time calling it.
2023-03-29 17:21:23,894:Retrying:WARNING:Finished call to '__main__.useit' after 1.010(s), this was the 2nd time calling it.
2023-03-29 17:21:25,896:Retrying:WARNING:Starting call to '__main__.useit', this is the 3rd time calling it.
2023-03-29 17:21:25,899:Retrying:WARNING:Finished call to '__main__.useit' after 3.015(s), this was the 3rd time calling it.

retrying failed ValueError('detailed reason')

Encore une fois, avec KeyError au lieu de ValueError:

try:
    useit(mock.MagicMock(side_effect=[KeyError("detailed reason")]*3))
except Exception as exc:
    print("retrying failed", repr(exc))

Le résultat:

2023-03-29 17:21:37,345:Retrying:WARNING:Starting call to '__main__.useit', this is the 1st time calling it.

retrying failed KeyError('detailed reason')

Séparez le décorateur du contrôleur

Souvent, des paramètres de nouvelle tentative similaires sont nécessaires à plusieurs reprises. Dans ces cas, il est préférable de créer un réessayer le contrôleur avec les paramètres :

my_retryer = tenacity.Retrying(
    stop=tenacity.stop_after_attempt(3),
    after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
    before=tenacity.before_log(TENACITY_LOGGER, logging.WARNING),
    retry=tenacity.retry_if_exception_type(ValueError),
    wait=tenacity.wait_incrementing(1, 10, 2),
    reraise=True
)

Décorez la fonction avec le contrôleur de nouvelle tentative :

@my_retryer.wraps
def useit(a_thing):
    value = a_thing()
    print("the value is", value)

Exécuter:

>>> useit(mock.MagicMock(side_effect=[ValueError(), 5]))

2023-03-29 17:29:25,656:Retrying:WARNING:Starting call to '__main__.useit', this is the 1st time calling it.
2023-03-29 17:29:25,663:Retrying:WARNING:Finished call to '__main__.useit' after 0.008(s), this was the 1st time calling it.
2023-03-29 17:29:26,667:Retrying:WARNING:Starting call to '__main__.useit', this is the 2nd time calling it.

the value is 5

Cela vous permet de rassembler les statistiques du dernier appel :

>>> my_retryer.statistics

{'start_time': 26782.847558759,
 'attempt_number': 2,
 'idle_for': 1.0,
 'delay_since_first_attempt': 0.0075125470029888675}

Utilisez ces statistiques pour mettre à jour un registre de statistiques internes et les intégrer à votre infrastructure de surveillance.

Prolonger la ténacité

La plupart des arguments du décorateur sont des objets. Ces objets peuvent être des objets de sous-classes, permettant une extension profonde.

Par exemple, supposons que la suite de Fibonacci doive déterminer les temps d’attente. La torsion est que l’API pour demander le temps d’attente ne donne que le nombre de tentatives, donc la méthode itérative habituelle de calcul de Fibonacci n’est pas utile.

Une façon d’atteindre l’objectif est d’utiliser le formule fermée:

Une astuce peu connue consiste à sauter la soustraction en faveur de l’arrondi à l’entier le plus proche :

Ce qui se traduit en Python par :

int(((1 + sqrt(5))/2)**n / sqrt(5) + 0.5)

Cela peut être utilisé directement dans une fonction Python :

from math import sqrt

def fib(n):
    return int(((1 + sqrt(5))/2)**n / sqrt(5) + 0.5)

La suite de Fibonacci compte à partir de 0 tandis que les numéros de tentative commencent à 1donc un wait la fonction doit compenser cela:

def wait_fib(rcs):
    return fib(rcs.attempt_number - 1)

La fonction peut être passée directement en tant que wait paramètre:

@tenacity.retry(
    stop=tenacity.stop_after_attempt(7),
    after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
    wait=wait_fib,
)
def useit(thing):
    print("value is", thing())
try:
    useit(mock.MagicMock(side_effect=[tenacity.TryAgain()] * 7))
except Exception as exc:
    pass

Essaye le:

2023-03-29 18:03:52,783:Retrying:WARNING:Finished call to '__main__.useit' after 0.000(s), this was the 1st time calling it.
2023-03-29 18:03:52,787:Retrying:WARNING:Finished call to '__main__.useit' after 0.004(s), this was the 2nd time calling it.
2023-03-29 18:03:53,789:Retrying:WARNING:Finished call to '__main__.useit' after 1.006(s), this was the 3rd time calling it.
2023-03-29 18:03:54,793:Retrying:WARNING:Finished call to '__main__.useit' after 2.009(s), this was the 4th time calling it.
2023-03-29 18:03:56,797:Retrying:WARNING:Finished call to '__main__.useit' after 4.014(s), this was the 5th time calling it.
2023-03-29 18:03:59,800:Retrying:WARNING:Finished call to '__main__.useit' after 7.017(s), this was the 6th time calling it.
2023-03-29 18:04:04,806:Retrying:WARNING:Finished call to '__main__.useit' after 12.023(s), this was the 7th time calling it.

Soustrayez les nombres suivants du temps “après” et du tour pour voir la suite de Fibonacci :

intervals = [
    0.000,
    0.004,
    1.006,
    2.009,
    4.014,
    7.017,
    12.023,
]
for x, y in zip(intervals[:-1], intervals[1:]):
    print(int(y-x), end=" ")

Est-ce que ça marche? Oui, exactement comme prévu :

0 1 1 2 3 5 

Conclure

L’écriture d’un code de nouvelle tentative ad hoc peut être une distraction amusante. Pour le code de production, un meilleur choix est une bibliothèque éprouvée comme tenacity. Le tenacity La bibliothèque est configurable et extensible, et elle répondra probablement à vos besoins.