Recherche de site Web

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

Utilisez les bibliothèques Tenacity et Mock pour trouver les bugs cachés au plus profond de votre code.

Parfois, une fonction est appelée avec de mauvaises entrées ou dans un mauvais état du programme, elle échoue donc. Dans des langages comme Python, cela entraîne généralement 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 la mise en cache. En théorie, le code et le nettoyeur pourraient soigneusement s'entendre sur la méthodologie de nettoyage pour empêcher le code de tenter 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 appropriées.

Plus fréquemment encore, 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 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é

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

Configurer la journalisation

Une fonctionnalité intégrée pratique de tenacity est la prise en charge de la journalisation. Avec la gestion des erreurs, consulter les détails du journal sur les nouvelles tentatives est inestimable.

Pour permettre aux exemples restants d'afficher les messages du 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 fait cela. Voici un exemple :

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 démontrer les fonctionnalités de ténacité, il est utile de pouvoir échouer plusieurs fois avant de finalement réussir. L'utilisation de 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 la simulation.

Avant de montrer la puissance de la ténacité, 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 à l'aide des sauvegardes ténacité.

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 consistait à réessayer 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.

La fonction useit n'a plus à se soucier de réessayer. Parfois, il est logique que la fonction considère toujours la réessaissabilité. Tenacity permet au code de déterminer lui-même la possibilité de réessayer en déclenchant 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)

Désormais, lors de l'appel de useit, il réessaye 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 ténacité prend en charge. 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
)

Il s'agit d'un exemple de décorateur plus réaliste avec des paramètres supplémentaires :

  • avant : journaliser avant d'appeler la fonction
  • retry : au lieu de simplement réessayer TryAgain, réessayez les exceptions avec les critères donnés
  • attendre : attendre entre les appels (ceci est particulièrement important si vous appelez un service)
  • reraise : si la nouvelle tentative échoue, relancez 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

Notez le délai entre la deuxième et la troisième ligne de journal. Cela dure 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 contrôleur de nouvelle tentative 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 interne et les intégrer à votre cadre de surveillance.

Étendre 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 séquence de Fibonacci détermine les temps d'attente. Le problème est que l'API permettant de demander le temps d'attente ne donne que le numéro de tentative, donc la méthode itérative habituelle de calcul de Fibonacci n'est pas utile.

Une façon d’atteindre cet objectif est d’utiliser la formule fermée :

Une astuce peu connue consiste à ignorer la soustraction au profit 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 séquence de Fibonacci compte à partir de 0 tandis que les numéros de tentatives commencent à 1, donc une fonction wait doit compenser cela :

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

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

@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 et du tour « après » pour voir la séquence 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

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

Articles connexes: