如何使用Tenacity在重试触发并执行成功时仅记录'retry_success'日志/发送Slack通知

我来帮你搞定这个问题～你当前的代码问题出在对Tenacity回调时机的理解上，咱们一步步理清楚：

问题根源拆解 #

1. before_sleep 回调的触发时机不对：这个函数只会在「函数执行失败，且即将进入睡眠等待下一次重试」时被调用，所以你写的else分支永远不会执行——函数成功的话根本不会走到这个回调环节。
2. after 回调缺少重试判断：你的原after逻辑没有区分「第一次就成功」和「重试后成功」的情况，而且不需要手动返回结果（Tenacity会自动处理结果传播）。

修正后的解决方案 #

我们只需要调整after回调的逻辑，同时保留before_sleep的重试警告逻辑，就能实现「仅当经历至少一次重试后成功时，记录特定日志」的需求：

import random
import logging
from tenacity import retry, wait_fixed, stop_after_attempt, RetryCallState, retry_if_exception_type

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def my_before_sleep(retry_state: RetryCallState):
    # 这个回调只在「失败后准备重试」时触发，所以直接处理重试警告即可
    logging.warning(
        f"Retrying '{retry_state.fn.__name__}' for the {retry_state.attempt_number} time "
        f"after {retry_state.outcome.exception()}..."
    )

def my_after_execute(retry_state: RetryCallState):
    # 两个关键判断：1. 最终执行成功；2. 确实经历过重试（attempt_number > 1）
    if not retry_state.outcome.failed and retry_state.attempt_number > 1:
        logging.info(
            f"'{retry_state.fn.__name__}' completed successfully after retries "
            f"(total attempts: {retry_state.attempt_number}). Result: {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!"

# 测试场景1：大概率会触发重试（prob=0.5，有50%概率第一次失败）
logging.info("Test 1: Should fail initially and then succeed with retries")
might_fail_function(0.5)

logging.info('-' * 50)

# 测试场景2：第一次就成功的情况（prob=0.1，只有10%概率失败）
logging.info("Test 2: Should succeed on first attempt (no retry log)")
might_fail_function(0.1)

关键逻辑说明 #

• my_before_sleep：保持原有的重试警告逻辑，因为它确实只在重试前触发，完美匹配需求。
• my_after_execute：
  • not retry_state.outcome.failed：确保最终执行是成功的
  • retry_state.attempt_number > 1：确保确实进行过至少一次重试（如果是1，说明第一次就成功，不需要打“重试后成功”的日志）
• 去掉了不必要的return语句，Tenacity会自动把最终结果传播出去，不需要手动处理。

测试效果 #

当函数经历重试后成功时，你会看到类似这样的日志：

2025-06-29 13:00:00,000 - INFO - Test 1: Should fail initially and then succeed with retries
2025-06-29 13:00:00,001 - INFO - Function failed on this attempt.
2025-06-29 13:00:00,001 - INFO - current prob: 0.3
2025-06-29 13:00:00,002 - WARNING - Retrying 'might_fail_function' for the 1 time after Simulated failure...
2025-06-29 13:00:01,003 - INFO - Function succeeded on this attempt.
2025-06-29 13:00:01,003 - INFO - current prob: 0.8
2025-06-29 13:00:01,004 - INFO - 'might_fail_function' completed successfully after retries (total attempts: 2). Result: Operation Completed!

而如果第一次就成功，只会看到函数本身的成功日志，不会触发“重试后成功”的特定日志。

内容来源于stack exchange