tenacity
tenacity 9.0.0
더 나은 경험을 위해서는 Tenacity 문서 를 참조하세요 .
Tenacity는 Python으로 작성된 Apache 2.0 라이선스 범용 재시도 라이브러리로, 거의 모든 것에 재시도 동작을 추가하는 작업을 단순화합니다. 이는 안타깝게도 더 이상 유지되지 않는 재시도 포크에서 발생합니다. Tenacity는 재시도와 호환되는 API는 아니지만 중요한 새 기능을 추가하고 오랫동안 지속된 여러 버그를 수정합니다.
가장 간단한 사용 사례는 값이 반환될 때까지 예외가 발생할 때마다 불안정한 함수를 다시 시도하는 것입니다.
.. 테스트코드::
무작위로 가져오기
Tenacity 가져오기 재시도에서
@다시 해 보다
def do_something_unreliable():
무작위.randint(0, 10) > 1인 경우:
raise IOError("깨진 소스, 모든 것이 엉망입니다!!!111one")
또 다른:
return "소스 정말 맛있어요!"
인쇄(do_something_unreliable())
.. 테스트 출력:: :숨다: 굉장한 소스!
.. 토트리::
:숨겨진:
:최대 깊이: 2
변경 내역
API
Tenacity 를 설치하려면 다음을 수행하세요.
$ pip install tenacity .. 테스트 설정::
수입 로깅
#
# 다음 가져오기는 데모 편의를 위해서만 사용됩니다.
# 프로덕션 코드는 항상 필요한 이름을 명시적으로 가져와야 합니다.
#
끈기 수입에서 *
클래스 MyException(예외):
통과하다
위에서 본 것처럼 기본 동작은 예외가 발생할 때 기다리지 않고 영원히 재시도하는 것입니다.
.. 테스트 코드::
@다시 해 보다
def never_gonna_give_you_up():
print("예외를 무시하고 영원히 재시도하세요. 재시도 사이에 기다리지 마세요.")
예외 발생
조금 덜 끈기를 갖고 포기하기 전의 시도 횟수 등 몇 가지 경계를 설정해 봅시다.
.. 테스트 코드::
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("7번 시도 후 중지됨")
예외 발생
하루 종일 시간이 없으므로 재시도해야 하는 시간에 대한 경계를 설정해 보겠습니다.
.. 테스트코드::
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("10초 후에 멈춥니다.")
예외 발생
기한이 촉박하고 지연 시간을 초과하는 것이 허용되지 않는 경우 지연을 초과하기 전에 한 번만 재시도를 포기할 수 있습니다.
.. 테스트코드::
@retry(중지=stop_before_delay(10))
def stop_before_10_s():
print("10초 전에 1번의 시도를 중지합니다.")
예외 발생
|를 사용하여 여러 중지 조건을 결합할 수 있습니다. 연산자:
.. 테스트 코드::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("10초 또는 5번의 재시도 후 중지")
예외 발생
대부분의 경우 가능한 한 빨리 폴링되는 것을 좋아하지 않으므로 재시도 사이에 2초만 기다려 보겠습니다.
.. 테스트 코드::
@재시도(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))
def wait_exponential_1():
print("각 재시도 사이에 4초부터 시작하여 2^x * 1초를 기다린 후 최대 10초를 기다린 후 10초를 기다립니다.")
예외 발생
또한 분산 서비스 및 기타 원격 엔드포인트를 재시도할 때 고정 대기와 지터(무리를 방지하기 위해)를 결합하는 것도 어렵습니다.
.. 테스트 코드::
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("최소 3초를 기다리고 최대 2초의 무작위 지연을 추가하세요.")
예외 발생
여러 프로세스가 공유 리소스를 놓고 경합할 때 지터를 기하급수적으로 늘리면 충돌을 최소화하는 데 도움이 됩니다.
.. 테스트 코드::
@retry(wait=wait_random_exponential(승수=1, 최대=60))
def wait_exponential_jitter():
print("범위가 60초에 도달할 때까지 각 재시도 사이에 무작위로 최대 2^x * 1초를 기다린 다음, 그 이후에는 무작위로 최대 60초를 기다립니다.")
예외 발생
때로는 백오프 체인을 구축해야 하는 경우도 있습니다.
.. 테스트코드::
@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
[범위(2) 내 i에 대한 wait_fixed(7)] +
[wait_fixed(9)]))
def wait_fixed_chained():
print("3회 시도는 3초, 다음 2회 시도는 7초, 그 이후 모든 시도는 9초를 기다립니다.")
예외 발생
여기의 경우처럼 특정 또는 일반 예외를 발생시키는 재시도를 처리하기 위한 몇 가지 옵션이 있습니다.
.. 테스트 코드::
클래스 ClientError(예외):
"""클라이언트 오류가 발생했습니다."""
@retry(재시도=retry_if_Exception_type(IOError))
def might_io_error():
print("IOError가 발생하면 기다리지 않고 계속 재시도하고 다른 오류를 발생시킵니다.")
예외 발생
@retry(재시도=retry_if_not_Exception_type(클라이언트 오류))
def might_client_error():
print("ClientError 이외의 오류가 발생하면 기다리지 않고 영원히 다시 시도하세요. 즉시 ClientError를 발생시키세요.")
예외 발생
또한 함수의 결과를 사용하여 재시도 동작을 변경할 수도 있습니다.
.. 테스트 코드::
def is_none_p(값):
"""값이 None이면 True를 반환합니다."""
반환 값은 없음입니다.
@retry(재시도=retry_if_result(is_none_p))
def might_return_none():
print("반환값이 None이면 기다리지 않고 다시 시도하세요.")
다음 방법도 참조하세요.
.. 테스트 코드::
재시도_if_예외
retry_if_Exception_type
retry_if_not_Exception_type
retry_unless_Exception_type
재시도_if_결과
retry_if_not_result
retry_if_Exception_message
retry_if_not_Exception_message
재시도_아무거나
재시도_모두
또한 여러 조건을 결합할 수도 있습니다.
.. 테스트 코드::
def is_none_p(값):
"""값이 None이면 True를 반환합니다."""
반환 값은 없음입니다.
@retry(retry=(retry_if_result(is_none_p) | retry_if_Exception_type()))
def might_return_none():
print("반환 값이 None인 경우 대기 없이 예외를 무시하고 영원히 재시도합니다.")
중지, 대기 등의 조합도 지원되므로 자유롭게 혼합하고 일치시킬 수 있습니다.
TryAgain 예외를 발생시켜 언제든지 명시적으로 재시도할 수도 있습니다.
.. 테스트 코드::
@다시 해 보다
def do_something():
결과 =some_else()
결과 == 23인 경우:
TryAgain을 올리세요
일반적으로 함수가 마지막에 실패하면(설정에 따라 다시 시도되지 않을 경우) RetryError가 발생합니다. 코드에서 발생한 예외는 스택 추적 중간 에 표시됩니다.
스택 추적이 끝날 때(가장 눈에 띄는 위치) 코드에서 발생한 예외를 확인하려면 reraise=True를 설정할 수 있습니다.
.. 테스트 코드::
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
노력하다:
raise_my_Exception()
MyException 제외:
# 재시도 시간이 초과되었습니다
통과하다
before 콜백 함수를 사용하여 함수 호출을 시도하기 전에 작업을 실행할 수 있습니다.
.. 테스트코드::
수입 로깅
수입 시스템
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
로거 = 로깅.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger,logging.DEBUG))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
같은 맥락에서 호출이 실패한 후에도 실행할 수 있습니다.
.. 테스트코드::
수입 로깅
수입 시스템
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
로거 = 로깅.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(로거, 로깅.DEBUG))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
재시도할 실패만 기록하는 것도 가능합니다. 일반적으로 재시도는 대기 간격 후에 발생하므로 키워드 인수는 before_sleep 이라고 합니다.
.. 테스트 코드::
수입 로깅
수입 시스템
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
로거 = 로깅.getLogger(__name__)
@retry(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(로거, 로깅.DEBUG))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
함수에 연결된 통계 속성을 사용하여 함수에 대한 재시도에 대한 통계에 액세스할 수 있습니다.
.. 테스트 코드::
@retry(stop=stop_after_attempt(3))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
노력하다:
raise_my_Exception()
예외:
통과하다
인쇄(raise_my_Exception.statistics)
.. 테스트 출력:: :숨다: ...
자신만의 콜백을 정의할 수도 있습니다. 콜백은 현재 재시도 호출에 대한 모든 정보를 포함하는 retry_state 라는 매개변수 하나를 허용해야 합니다.
예를 들어, 모든 재시도가 실패한 후 예외를 발생시키지 않고 사용자 정의 콜백 함수를 호출할 수 있습니다(또는 실제로 다시 발생시키거나 다른 작업을 수행할 수도 있습니다).
.. 테스트코드::
def return_last_value(재시도_상태):
"""마지막 통화 시도 결과를 반환합니다"""
retry_state.outcome.result()를 반환합니다.
def is_false(값):
"""값이 False이면 True를 반환합니다."""
반환 값은 False입니다.
# 다른 결과를 얻기 위해 3번 시도한 후에는 False를 반환합니다.
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
재시도=retry_if_result(is_false))
def 결국_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)
로거 = 로깅.getLogger(__name__)
def my_before_sleep(retry_state):
retry_state.attempt_number < 1인 경우:
로그레벨 = 로깅.INFO
또 다른:
loglevel = 로깅.경고
로거.로그(
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)
def raise_my_Exception():
MyException("실패")을 발생시키세요.
노력하다:
raise_my_Exception()
RetryError 제외:
통과하다
래핑된 함수에 첨부된 retry_with 함수를 사용하여 재시도 데코레이터를 호출할 때 필요에 따라 재시도 데코레이터의 인수를 변경할 수 있습니다.
.. 테스트 코드::
@retry(stop=stop_after_attempt(3))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
노력하다:
raise_my_Exception.retry_with(stop=stop_after_attempt(4))()
예외:
통과하다
인쇄(raise_my_Exception.statistics)
.. 테스트 출력:: :숨다: ...
변수를 사용하여 재시도 매개변수를 설정하려는 경우 retry 데코레이터를 사용할 필요가 없습니다. 대신 Retrying을 직접 사용할 수 있습니다.
.. 테스트코드::
def never_good_enough(arg1):
raise Exception('잘못된 인수: {}'.format(arg1))
def try_never_good_enough(max_attempts=3):
재시도자 = 재시도 중(stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, '정말로 시도합니다')
불필요한 대기 시간을 피하기 위해 테스트처럼 장식된 함수의 동작을 일시적으로 변경할 수도 있습니다. 함수에 첨부된 재시도 속성을 수정/패치할 수 있습니다. 이는 쓰기 전용 속성이므로 통계는 함수 통계 속성에서 읽어야 합니다.
.. 테스트 코드::
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def raise_my_Exception():
MyException("실패")을 발생시키세요.
Unittest import mock에서
mock.patch.object(raise_my_Exception.retry, "wait", wait_fixed(0))를 사용하여:
노력하다:
raise_my_Exception()
예외:
통과하다
인쇄(raise_my_Exception.statistics)
.. 테스트 출력:: :숨다: ...
Tenacity를 사용하면 코드 블록을 격리된 함수로 래핑할 필요 없이 코드 블록을 다시 시도할 수 있습니다. 이를 통해 컨텍스트를 공유하면서 실패한 블록을 쉽게 격리할 수 있습니다. 비결은 for 루프와 컨텍스트 관리자를 결합하는 것입니다.
.. 테스트코드::
from tenacity import 재시도, RetryError, stop_after_attempt
노력하다:
재시도 시도(stop=stop_after_attempt(3)):
시도:
raise Exception('내 코드가 실패했습니다!')
RetryError 제외:
통과하다
Retrying 객체를 구성하여 재시도 정책의 모든 세부 사항을 구성할 수 있습니다.
비동기 코드를 사용하면 AsyncRetrying을 사용할 수 있습니다.
.. 테스트코드::
Tenacity import AsyncRetrying, RetryError, stop_after_attempt에서
비동기 정의 함수():
노력하다:
AsyncRetrying(stop=stop_after_attempt(3)) 시도에 대한 비동기:
시도:
raise Exception('내 코드가 실패했습니다!')
RetryError 제외:
통과하다
두 경우 모두 retry_if_result 와 같은 재시도 전략에서 사용할 수 있도록 결과를 시도로 설정할 수 있습니다. 이는 retry_state 속성에 액세스하여 수행할 수 있습니다.
.. 테스트 코드::
Tenacity import AsyncRetrying, retry_if_result에서
비동기 정의 함수():
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
