无法记录 "retry_success", 只有在触发重试时才会记录
作者: vasu228114创建于 2025年6月29日更新于 2025年6月30日
问题: 尝试在重试触发时生成成功消息。能够正确生成警告,但如果发生重试,则不生成成功消息。 使用以下代码:
import random
import logging
from tenacity import retry, wait_fixed, stop_after_attempt, before_sleep, RetryCallState, retry_if_exception_type
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def my_before_sleep(retry_state: RetryCallState):
if retry_state.outcome.failed:
# 如果失败,请在此处不要做任何特殊处理;before_sleep 会处理警告。
logging.warning(
f"重试 '{retry_state.fn.__name__}' 为 {retry_state.attempt_number} 次,"
f"在 {retry_state.outcome.exception()} 后..."
)
return retry_state.outcome.result # 返回 tenacity 处理的异常
else:
# 此 ELSE 不能正常工作,如预期
logging.info(
f"'{retry_state.fn.__name__}' 在重试后成功完成(尝试 {retry_state.attempt_number})。结果:{retry_state.outcome.result}"
)
def my_after_execute(retry_state: RetryCallState):
"""
每次尝试后执行的回调函数,无论成功与否。
检查函数是否成功,以及是否之前发生过重试。
"""
if not retry_state.outcome.failed:
# 如果上述 ELSE 不能正常工作,即使此 if 条件也不能正常工作。
logging.info(
f"'{retry_state.fn.__name__}' 在重试后成功完成(尝试 {retry_state.attempt_number})。结果:{retry_state.outcome.result}"
)
return retry_state.outcome.result # 返回 tenacity 传播的实际结果
@retry(
wait=wait_fixed(1),
stop=stop_after_attempt(3),
retry=retry_if_exception_type(ValueError),
before_sleep=my_before_sleep,
after=my_after_execute,
reraise=True
)
def might_fail_function(prob):
# 模拟成功或失败
current_prob = random.random()
if current_prob < prob:
logging.info("Function failed on this attempt.")
logging.info(f"current prob: {current_prob}")
raise ValueError("Simulated failure")
else:
logging.info("Function succeeded on this attempt.")
logging.info(f"current prob: {current_prob}")
return "Operation Completed!"
logging.info("Attempt 1: Should fail initially and then succeed with retries")
# 在第 1 次尝试中失败,然后成功(此部分不工作)
might_fail_function(0.5)
logging.info('-' * 50)内容来源: jd/tenacity