tenacity
tenacity 9.0.0
Пожалуйста, обратитесь к документации Tenacity для лучшего опыта.
Tenacity — это лицензированная Apache 2.0 библиотека повторных попыток общего назначения, написанная на Python для упрощения задачи добавления поведения повторной попытки практически к чему угодно. Это происходит из-за ответвления повторной попытки, которое, к сожалению, больше не поддерживается. Tenacity не совместим с повторной попыткой API, но добавляет существенные новые функции и исправляет ряд давних ошибок.
Самый простой вариант использования — повторная попытка нестабильной функции всякий раз, когда возникает исключение, до тех пор, пока не будет возвращено значение.
.. тестовый код::
импортировать случайный
из-за упорства повторить попытку импорта
@retry
защита do_something_unreliable():
если random.randint(0, 10) > 1:
поднять IOError("Соус сломан, всё залито!!!111one")
еще:
ответ "Потрясающий соус!"
печать (do_something_unreliable())
.. тестовый вывод:: :скрывать: Обалденный соус!
.. toctree::
:скрытый:
:максглубина: 2
журнал изменений
API
Чтобы установить Tenacity , просто:
$ pip install tenacity .. тестовая установка::
журнал импорта
#
# Обратите внимание, что следующий импорт используется только для удобства демонстрации.
# Производственный код всегда должен явно импортировать нужные ему имена.
#
от импорта цепкости *
класс MyException (Исключение):
проходить
Как вы видели выше, поведением по умолчанию является постоянное повторение попыток, не дожидаясь возникновения исключения.
.. тестовый код::
@retry
защита Never_gonna_give_you_up():
print("Повторять попытки всегда, игнорируя исключения, не ждите между повторами")
поднять исключение
Давайте будем чуть менее настойчивыми и установим некоторые границы, например количество попыток, прежде чем сдаться.
.. тестовый код::
@retry(stop=stop_after_attempt(7))
защита stop_after_7_attempts():
print("Остановка после 7 попыток")
поднять исключение
У нас нет целого дня, поэтому давайте установим границу того, как долго мы должны повторять попытки.
.. тестовый код::
@retry(stop=stop_after_delay(10))
защита stop_after_10_s():
print("Остановка через 10 секунд")
поднять исключение
Если у вас сжатые сроки и превышение времени задержки недопустимо, вы можете отказаться от повторных попыток за одну попытку, прежде чем вы превысите задержку.
.. тестовый код::
@retry(stop=stop_before_delay(10))
защита stop_before_10_s():
print("Остановка 1 попытки до истечения 10 секунд")
поднять исключение
Вы можете объединить несколько условий остановки, используя | оператор:
.. тестовый код::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
защита stop_after_10_s_or_5_retries():
print("Остановка через 10 секунд или 5 попыток")
поднять исключение
Большинству вещей не нравится, когда опрос выполняется настолько быстро, насколько это возможно, поэтому давайте просто подождем 2 секунды между повторными попытками.
.. тестовый код::
@retry(wait=wait_fixed(2))
защита wait_2_s():
print("Подождите 2 секунды между попытками")
поднять исключение
Некоторые вещи работают лучше всего, если добавить немного случайности.
.. тестовый код::
@retry(wait=wait_random(мин=1, макс=2))
защита wait_random_1_to_2_s():
print("Между повторными попытками происходит произвольная пауза от 1 до 2 секунд")
поднять исключение
Опять же, трудно преодолеть экспоненциальную задержку при повторных попытках распределенных служб и других удаленных конечных точек.
.. тестовый код::
@retry(wait=wait_exponential(множитель=1, мин=4, макс=10))
защита wait_exponential_1():
print("Подождите 2^x * 1 секунда между каждой повторной попыткой, начиная с 4 секунд, затем до 10 секунд, затем еще 10 секунд")
поднять исключение
С другой стороны, также трудно победить сочетание фиксированного ожидания и дрожания (чтобы избежать громовых стад) при повторных попытках распределенных служб и других удаленных конечных точек.
.. тестовый код::
@retry(wait=wait_fixed(3) + wait_random(0, 2))
защита wait_fixed_jitter():
print("Подождите не менее 3 секунд и добавьте до 2 секунд случайной задержки")
поднять исключение
Когда несколько процессов конкурируют за общий ресурс, экспоненциально увеличивающийся джиттер помогает свести к минимуму коллизии.
.. тестовый код::
@retry(wait=wait_random_exponential(множитель=1, максимум=60))
защита wait_exponential_jitter():
print("Произвольно ждать до 2^x * 1 секунды между каждой повторной попыткой, пока диапазон не достигнет 60 секунд, затем случайным образом до 60 секунд после этого")
поднять исключение
Иногда необходимо выстроить цепочку откатов.
.. тестовый код::
@retry(wait=wait_chain(*[wait_fixed(3) для i в диапазоне(3)] +
[wait_fixed(7) для i в диапазоне(2)] +
[wait_fixed(9)]))
защита wait_fixed_chained():
print("Подождите 3 секунды для 3 попыток, 7 секунд для следующих 2 попыток и 9 секунд для всех последующих попыток")
поднять исключение
У нас есть несколько вариантов борьбы с повторными попытками, которые вызывают конкретные или общие исключения, как в данном случае.
.. тестовый код::
класс ClientError (Исключение):
"""Какая-то ошибка клиента."""
@retry(retry=retry_if_Exception_type(IOError))
защита may_io_error():
print("Повторять попытку бесконечно, без ожидания, если возникает ошибка IOError, вызывать любые другие ошибки")
поднять исключение
@retry(retry=retry_if_not_Exception_type(ClientError))
защита may_client_error():
print("Повторять попытку бесконечно, без ожидания, если возникает какая-либо ошибка, кроме ClientError. Немедленно вызвать ClientError.")
поднять исключение
Мы также можем использовать результат функции, чтобы изменить поведение повторной попытки.
.. тестовый код::
защита is_none_p (значение):
"""Верните True, если значение равно None"""
возвращаемое значение: Нет
@retry(retry=retry_if_result(is_none_p))
защита may_return_none():
print("Повторить попытку без ожидания, если возвращаемое значение равно None")
См. также эти методы:
.. тестовый код::
retry_if_Exception
retry_if_Exception_type
retry_if_not_Exception_type
retry_unless_Exception_type
retry_if_result
retry_if_not_result
retry_if_Exception_message
retry_if_not_Exception_message
retry_any
retry_all
Мы также можем объединить несколько условий:
.. тестовый код::
защита is_none_p (значение):
"""Верните True, если значение равно None"""
возвращаемое значение: Нет
@retry(retry=(retry_if_result(is_none_p) | retry_if_Exception_type()))
защита may_return_none():
print("Повторить попытку навсегда игнорировать исключения без ожидания, если возвращаемое значение равно None")
Также поддерживается любая комбинация остановки, ожидания и т. д., что дает вам свободу смешивать и сочетать.
Также возможно повторить попытку в любое время, вызвав исключение TryAgain:
.. тестовый код::
@retry
защита do_something():
результат = что-то_еще()
если результат == 23:
поднять TryAgain
Обычно, когда ваша функция завершается с ошибкой в последний раз (и не будет повторяться в зависимости от ваших настроек), возникает RetryError. Исключение, с которым столкнулся ваш код, будет показано где-то в середине трассировки стека.
Если вы предпочитаете видеть исключение, с которым столкнулся ваш код, в конце трассировки стека (где оно наиболее заметно), вы можете установить reraise=True.
.. тестовый код::
@retry(reraise=True, stop=stop_after_attempt(3))
защита raise_my_Exception():
поднять MyException («Ошибка»)
пытаться:
поднять_мое_исключение()
кроме MyException:
# время ожидания повторной попытки
проходить
Можно выполнить действие перед любой попыткой вызова функции, используя функцию обратного вызова before:
.. тестовый код::
журнал импорта
импортировать систему
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
защита raise_my_Exception():
поднять MyException («Ошибка»)
В том же духе можно выполнить после неудачного вызова:
.. тестовый код::
журнал импорта
импортировать систему
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
защита raise_my_Exception():
поднять MyException («Ошибка»)
Также возможно регистрировать только те сбои, которые будут повторены. Обычно повторные попытки происходят после интервала ожидания, поэтому аргумент ключевого слова вызывается before_sleep :
.. тестовый код::
журнал импорта
импортировать систему
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(регистратор, ведение журнала.DEBUG))
защита raise_my_Exception():
поднять MyException («Ошибка»)
Вы можете получить доступ к статистике повторных попыток, выполненных с помощью функции, используя атрибут статистики, прикрепленный к функции:
.. тестовый код::
@retry(stop=stop_after_attempt(3))
защита raise_my_Exception():
поднять MyException («Ошибка»)
пытаться:
поднять_мое_исключение()
кроме Исключения:
проходить
печать (raise_my_Exception.statistics)
.. тестовый вывод:: :скрывать: ...
Вы также можете определить свои собственные обратные вызовы. Обратный вызов должен принимать один параметр с именем retry_state , который содержит всю информацию о текущем повторном вызове.
Например, вы можете вызвать пользовательскую функцию обратного вызова после неудачных попыток, не вызывая исключения (или вы можете повторно вызвать или сделать что-нибудь действительно)
.. тестовый код::
защита return_last_value(retry_state):
"""возвращает результат последней попытки вызова"""
вернуть retry_state.outcome.result()
защита is_false (значение):
"""Верните True, если значение равно False"""
возвращаемое значение ложно
# вернет False после 3-х попыток получить другой результат
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
повтор=retry_if_result(is_false))
защита в конечном итоге_return_false():
вернуть ложь
Аргумент retry_state является объектом класса :class:`~tenacity.RetryCallState` .
Также возможно определить собственные обратные вызовы для других аргументов ключевого слова.
.. функция:: my_stop(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове :return: следует ли прекратить повторную попытку :rtype: бул
.. функция:: my_wait(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове :return: количество секунд ожидания перед следующей повторной попыткой :rtype: плавающий
.. функция:: my_retry(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове :return: следует ли продолжать повторную попытку :rtype: бул
.. функция:: my_before(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове
.. функция:: my_after(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове
.. функция:: my_before_sleep(retry_state) :param RetryCallState retry_state: информация о текущем повторном вызове
Вот пример пользовательской функции before_sleep :
.. тестовый код::
журнал импорта
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
защита my_before_sleep(retry_state):
если retry_state.attempt_number < 1:
уровень журнала = logging.INFO
еще:
уровень журнала = ведение журнала. ПРЕДУПРЕЖДЕНИЕ
logger.log(
loglevel, 'Повторная попытка %s: попытка %s закончилась: %s',
retry_state.fn, retry_state.attempt_number, retry_state.outcome)
@retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
защита поднять_my_Exception():
поднять MyException («Ошибка»)
пытаться:
поднять_мое_исключение()
кроме RetryError:
проходить
Вы можете изменить аргументы декоратора повтора по мере необходимости при его вызове, используя функцию retry_with, прикрепленную к обернутой функции:
.. тестовый код::
@retry(stop=stop_after_attempt(3))
защита raise_my_Exception():
поднять MyException («Ошибка»)
пытаться:
поднять_my_Exception.retry_with(stop=stop_after_attempt(4))()
кроме Исключения:
проходить
печать (raise_my_Exception.statistics)
.. тестовый вывод:: :скрывать: ...
Если вы хотите использовать переменные для настройки параметров повтора, вам не обязательно использовать декоратор повтора — вместо этого вы можете использовать Retrying напрямую:
.. тестовый код::
защита Never_good_enough(arg1):
поднять исключение('Неверный аргумент: {}'.format(arg1))
защита try_never_good_enough(max_attempts=3):
retryer = Повторная попытка (stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, 'Я действительно стараюсь')
Вы также можете временно изменить поведение декорированной функции, например, в тестах, чтобы избежать ненужного времени ожидания. Вы можете изменить/исправить атрибут повтора, прикрепленный к функции. Имейте в виду, что этот атрибут доступен только для записи, статистику следует считывать из атрибута статистики функции.
.. тестовый код::
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
защита raise_my_Exception():
поднять MyException («Ошибка»)
из макета импорта unittest
с помощьюock.patch.object(raise_my_Exception.retry, "wait", wait_fixed(0)):
пытаться:
поднять_мое_исключение()
кроме Исключения:
проходить
печать (raise_my_Exception.statistics)
.. тестовый вывод:: :скрывать: ...
Tenacity позволяет вам повторить блок кода без необходимости заключать его в изолированную функцию. Это позволяет легко изолировать неисправный блок при совместном использовании контекста. Хитрость заключается в том, чтобы объединить цикл for и менеджер контекста.
.. тестовый код::
из упорного импорта Retrying, RetryError, stop_after_attempt
пытаться:
для попытки повторной попытки (stop=stop_after_attempt(3)):
с попыткой:
поднять исключение («Мой код не работает!»)
кроме RetryError:
проходить
Вы можете настроить все детали политики повторных попыток, настроив объект «Повторная попытка».
С асинхронным кодом вы можете использовать AsyncRetrying.
.. тестовый код::
из упорного импорта AsyncRetrying, RetryError, stop_after_attempt
асинхронная функция определения():
пытаться:
асинхронная попытка в AsyncRetrying(stop=stop_after_attempt(3)):
с попыткой:
поднять исключение («Мой код не работает!»)
кроме RetryError:
проходить
В обоих случаях вы можете установить результат попытки, чтобы он был доступен в стратегиях повтора, таких как retry_if_result . Это можно сделать, обратившись к свойству retry_state :
.. тестовый код::
из упорного импорта AsyncRetrying, retry_if_result
асинхронная функция определения():
async для попытки в AsyncRetrying(retry=retry_if_result(lambda x: x <3)):
с попыткой:
result = 1 # Некоторые сложные вычисления, вызов функций и т. д.
если не попытка.retry_state.outcome.failed:
попытка.retry_state.set_result(результат)
вернуть результат
Наконец, retry работает также с сопрограммами asyncio, Trio и Tornado (>= 4.5). Сон также выполняется асинхронно.
@ 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 )Вы даже можете использовать альтернативные циклы событий, такие как curio, передав правильную функцию сна:
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno используется для управления журналами изменений. Взгляните на их документацию по использованию.
Генерация документа автоматически скомпилирует журналы изменений. Вам просто нужно их добавить.
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
