pip install tenacity

from tenacity import retry, stop_after_attempt, wait_fixed, before_log

# 基礎(chǔ)重試結(jié)構(gòu)
@retry(
    stop=stop_after_attempt(3),      # 最多重試3次
    wait=wait_fixed(2),               # 每次間隔2秒
    before=before_log(logger, 'WARNING')  # 重試前記錄日志
)
def api_call():
    # 可能失敗的邏輯

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

# 配置日志記錄器
logger = logging.getLogger(__name__)

def custom_before_log(retry_state: RetryCallState):
    """自定義重試前日志鉤子"""
    if retry_state.attempt_number > 1:
        logger.warning(
            f"函數(shù) {retry_state.fn.__name__} 第{retry_state.attempt_number}次重試"
        )

@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    before=custom_before_log  # 掛載鉤子函數(shù)
)
def login(user, password):
    # 模擬登錄操作
    if random.random() > 0.4:
        raise ConnectionError("認證服務(wù)不可用")
    return "登錄成功"

import inspect
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_fixed

def retry_with_logging(stop_max=3, wait_seconds=2):
    """帶日志記錄的自定義重試裝飾器"""
    def decorator(func):
        @wraps(func)
        @retry(
            stop=stop_after_attempt(stop_max),
            wait=wait_fixed(wait_seconds)
        )
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                # 動態(tài)獲取函數(shù)參數(shù)
                sig = inspect.signature(func)
                bound_args = sig.bind(*args, **kwargs)
                args_str = ", ".join(
                    f"{k}={v!r}" for k,v in bound_args.arguments.items()
                )
                
                # 記錄詳細日志
                logger.warning(
                    f"函數(shù) {func.__name__}({args_str}) "
                    f"第{wrapper.retry_state.attempt_number}次失敗: {e}"
                )
                raise e
        return wrapper
    return decorator

# 使用示例
@retry_with_logging(stop_max=4, wait_seconds=3)
def process_data(data_id, priority='high'):
    # 數(shù)據(jù)處理邏輯
    if random.random() > 0.3:
        raise ResourceWarning("資源暫時不可用")

inspect.signature(func)  # 獲取函數(shù)簽名
sig.bind(*args, **kwargs)  # 綁定實際參數(shù)

wrapper.retry_state.attempt_number  # 當前重試次數(shù)

from tenacity import wait_exponential

@retry(wait=wait_exponential(multiplier=1, max=60))
def api_call():
    # 等待時間：1s → 2s → 4s → ... → 60s

from tenacity import retry_if_exception_type

@retry(retry=retry_if_exception_type((TimeoutError, ConnectionError)))

@retry(
    stop=(stop_after_attempt(5) | stop_after_delay(30)),  # 5次或30秒后停止
    wait=wait_random(min=1, max=10),                     # 隨機等待1-10秒
    after=release_resource  # 重試結(jié)束后釋放資源
)

def cleanup(retry_state):
    if retry_state.outcome.failed:
        close_db_connection()

from sentry_sdk import capture_message

def alert_on_retry(retry_state):
    if retry_state.attempt_number > 3:
        capture_message(f"高頻重試: {retry_state.fn.__name__}")

# 最佳實踐模板
@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(max=30),
    before=log_attempt_number,
    retry=retry_if_exception_type(TransientError)
)