tenacity Python

Введение

Tenacity — это библиотека для добавления повторных попыток выполнения любых задач.

Она является форком библиотеки retrying, которая больше не поддерживается.

Tenacity несовместима с API retrying, но добавляет существенные новые функции и исправляет ряд давних ошибок.

Распространяется по лицензии Apache 2.0.

Официальная документация

Установка

python -m pip install tenacity

Collecting tenacity Downloading tenacity-9.1.2-py3-none-any.whl.metadata (1.2 kB) Downloading tenacity-9.1.2-py3-none-any.whl (28 kB) Installing collected packages: tenacity Successfully installed tenacity-9.1.2

Примеры

Если декорировать функцию, которая просто печатает сообщение - эффекта от @retry не будет

from tenacity import retry @retry def my_func(): print("Executing") my_func()

Executing

Если добавить исключение - тогда вызов будет повторяться бесконечно.

from tenacity import retry @retry def my_func(): print("Executing") raise Exception("My exception") my_func()

Executing Executing Executing Executing Executing Executing Executing …

Можно импортировать более конкретную RetryError

from tenacity import retry, RetryError @retry def my_func(): print("Executing") raise RetryError

Executing Executing Executing Executing Executing Executing Executing …

Существует несколько приёмов, с помощью которых обычно ограничивают число попыток. Обсудим их в следующих параграфах.

stop_after_attempt

С помощью stop_after_attempt() можно указать максимальное число попыток

from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(2)) def my_func(): print("Executing") raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

Executing Executing RetryError[<Future at 0x223f4d4bb60 state=finished raised Exception>] Closing

В этот пример можно добавить обработку именно RetryError

from tenacity import retry, stop_after_attempt, RetryError @retry(stop=stop_after_attempt(2)) def my_func(): print("Executing") raise Exception("My exception") try: my_func() except RetryError as e: print("RetryError: ", e) except Exception as e: print("Exception", e) finally: print("Closing")

Executing Executing RetryError: RetryError[<Future at 0x15e04a3bce0 state=finished raised Exception>] Closing

stop_after_delay

С помощью stop_after_delay() можно указать максимальное время работы

from tenacity import retry, stop_after_delay @retry(stop=stop_after_delay(2)) def my_func(): print("Executing") raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

Executing Executing … Executing Executing RetryError[<Future at 0x223f4d4bb60 state=finished raised Exception>] Closing

wait_fixed

С помощью wait_fixed() можно указать задержку между повторами

from tenacity import retry, stop_after_attempt, wait_fixed @retry(stop=stop_after_attempt(2), wait=wait_fixed(1)) def my_func(): print("Executing") raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

Executing Executing RetryError[<Future at 0x223f4d4bb60 state=finished raised Exception>] Closing

Теперь между Executing и Executing проходит секунда.

wait_random

С помощью wait_random() можно добавить неопределённости в задержку между повторами

from tenacity import retry, stop_after_attempt, wait_random @retry(stop=stop_after_attempt(2), wait=wait_random(min=1, max=5)) def my_func(): print("Executing") raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

Executing Executing RetryError[<Future at 0x223f4d4bb60 state=finished raised Exception>] Closing

Теперь между Executing и Executing проходит от одной до пяти секунд.

retry_if_exception_type

С помощью retry_if_exception_type() можно добавить условие на перезапуск только при определённом исключении

from tenacity import retry, retry_if_exception_type, stop_after_attempt @retry( retry=retry_if_exception_type(ValueError), stop=stop_after_attempt(2) ) def my_func(var): print(f"Executing with {var = }") if var == 1: raise ValueError("My ValueError") else: raise Exception("My exception") try: my_func(1) except Exception as e: print(e) finally: print("Closing")

Executing with var = 1 Executing with var = 1 RetryError[<Future at 0x1d2fd6bbb00 state=finished raised ValueError>] Closing

# … try: my_func(2) except Exception as e: print(e) finally: print("Closing")

Executing with var = 2 My exception Closing

Добавим немного случайности. Пусть теперь перезапуск зависит от того какое значение у случайного числа.

from tenacity import retry, retry_if_exception_type, stop_after_attempt from random import randint @retry( retry=retry_if_exception_type(ValueError), stop=stop_after_attempt(5) ) def my_func(): var = randint(0, 9) print(f"Executing with {var = }") if var > 1: raise ValueError("My ValueError") else: raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

Executing with var = 2 Executing with var = 5 Executing with var = 5 Executing with var = 8 Executing with var = 0 My exception Closing

В вызове выше пяти перезапусков хватило, чтобы получить числе меньше или равное 1.

Но может быть и такой вариант

Executing with var = 2 Executing with var = 6 Executing with var = 8 Executing with var = 6 Executing with var = 4 RetryError[<Future at 0x2350d35bef0 state=finished raised ValueError>] Closing

retry_if_result

С помощью retry_if_result() можно добавить условие на перезапуск только при определённом возвращаемом значении.

В этом примере используется лямбда функция

from tenacity import retry, retry_if_result, stop_after_attempt @retry( retry=retry_if_result(lambda result: result is None), stop=stop_after_attempt(2) ) def my_func(var): print(f"Executing with {var = }") return None if var == 1 else var try: my_func(1) except Exception as e: print(e) finally: print("Closing") try: my_func(2) except Exception as e: print(e) finally: print("Closing")

Executing with var = 1 Executing with var = 1 RetryError[<Future at 0x2779661bef0 state=finished returned NoneType>] Closing Executing with var = 2 Closing

Логи

С помощью before_sleep можно организовать вывод логов перед каждым перезапуском.

import logging from tenacity import retry, stop_after_delay, wait_fixed logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) def log_before_retry(retry_state): loglevel = logging.INFO if retry_state.attempt_number < 4 else logging.WARNING logger.log( loglevel, "Retrying %s: wait: %ss,\nattempt %s ended with: %s", retry_state.fn, retry_state.upcoming_sleep, retry_state.attempt_number, retry_state.outcome, ) @retry( before_sleep=log_before_retry, reraise=True, stop=stop_after_delay(4), wait=wait_fixed(1) ) def my_func(): print(f"Executing") raise Exception("My exception") try: my_func() except Exception as e: print(e) finally: print("Closing")

INFO:__main__:Retrying <function my_func at 0x0000010BE68B9080>: wait: 1.0s, attempt 1 ended with: <Future at 0x10be68c02f0 state=finished raised Exception> Executing INFO:__main__:Retrying <function my_func at 0x0000010BE68B9080>: wait: 1.0s, attempt 2 ended with: <Future at 0x10be68c16a0 state=finished raised Exception> Executing INFO:__main__:Retrying <function my_func at 0x0000010BE68B9080>: wait: 1.0s, attempt 3 ended with: <Future at 0x10be68c1610 state=finished raised Exception> Executing Executing WARNING:__main__:Retrying <function my_func at 0x0000010BE68B9080>: wait: 1.0s, attempt 4 ended with: <Future at 0x10be68c16d0 state=finished raised Exception> Executing My exception Closing

asyncio

import asyncio from tenacity import retry, stop_after_attempt @retry( reraise=True, stop=stop_after_attempt(2) ) def my_async_func(): print(f"Executing") raise Exception("My exception") try: asyncio.run(my_async_func()) except Exception as e: print(e) finally: print("Closing")

Executing Executing My exception Closing

Retrying

С помощью Retrying можно перезапускать блок кода без использования функций.

from tenacity import Retrying, RetryError, stop_after_attempt try: for attempt in Retrying(stop=stop_after_attempt(2)): with attempt: print("Execute") print(f"Retry = {attempt.retry_state.attempt_number}") 1 / 0 except RetryError: print("Failed after retries.")

Execute Retry = 1 Execute Retry = 2 Failed after retries.

Автор статьи: Андрей Олегович