全栈开发

tenacity 重试库入门

Posted by

Demo 1: 基础入门

"""
demo1_basics.py - 感受tenacity的基本魔力
"""
from tenacity import retry, stop_after_attempt, wait_fixed
import time

# ========== 最简单的重试 ==========
@retry
def infinite_retry():
    print("我会一直重试...按Ctrl+C停止")
    raise Exception("永远失败")
    time.sleep(1)

# ========== 限制重试3次 ==========
@retry(stop=stop_after_attempt(3))
def try_3_times():
    print(f"[{time.strftime('%H:%M:%S')}] 尝试执行...")
    raise Exception("失败了")

# ========== 每次等待2秒 ==========
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def wait_2_seconds():
    print(f"[{time.strftime('%H:%M:%S')}] 执行")
    raise Exception("失败")

if __name__ == '__main__':
    print("=" * 40)
    print("测试1: 限制重试3次")
    print("=" * 40)
    try:
        try_3_times()
    except Exception as e:
        print(f"最终失败: {e}")
    
    print("\n" + "=" * 40)
    print("测试2: 每次等待2秒")
    print("=" * 40)
    try:
        wait_2_seconds()
    except Exception as e:
        print(f"最终失败: {e}")

Demo 2: 等待策略对比

"""
demo2_wait_strategies.py - 不同的等待策略效果
"""
from tenacity import (
    retry, stop_after_attempt, 
    wait_fixed, wait_random, wait_exponential
)
import time
import random

# ========== 固定等待 ==========
@retry(stop=stop_after_attempt(4), wait=wait_fixed(2))
def fixed_wait():
    print(f"[{time.strftime('%H:%M:%S')}] 固定等2秒")
    raise Exception("失败")

# ========== 随机等待 ==========
@retry(stop=stop_after_attempt(4), wait=wait_random(min=1, max=5))
def random_wait():
    wait_time = random.randint(1, 5)
    print(f"[{time.strftime('%H:%M:%S')}] 随机等待(模拟)")
    raise Exception("失败")

# ========== 指数退避(生产环境推荐) ==========
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30)
)
def exponential_wait():
    print(f"[{time.strftime('%H:%M:%S')}] 指数退避")
    raise Exception("失败")

# ========== 模拟真实场景 ==========
class APIClient:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=1, max=10)
    )
    def fetch_data(self):
        print(f"[{time.strftime('%H:%M:%S')}] 调用API...")
        if random.random() < 0.7:  # 70%失败率
            raise ConnectionError("网络超时")
        return {"status": "success"}

if __name__ == '__main__':
    print("1. 固定等待策略")
    print("-" * 30)
    try:
        fixed_wait()
    except:
        pass
    
    print("\n2. 指数退避策略(推荐)")
    print("-" * 30)
    try:
        exponential_wait()
    except:
        pass
    
    print("\n3. 模拟API调用")
    print("-" * 30)
    client = APIClient()
    try:
        result = client.fetch_data()
        print(f"成功: {result}")
    except Exception as e:
        print(f"失败: {e}")

Demo 3: 条件重试

"""
demo3_conditions.py - 智能判断何时重试
"""
from tenacity import (
    retry, stop_after_attempt,
    retry_if_exception_type, retry_if_result
)

# ========== 只重试网络错误 ==========
@retry(
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type((ConnectionError, TimeoutError))
)
def selective_retry():
    """网络错误重试，其他错误不重试"""
    import random
    error_type = random.choice(['network', 'data', 'success'])
    
    if error_type == 'network':
        raise ConnectionError("网络超时 -> 会重试")
    elif error_type == 'data':
        raise ValueError("数据格式错误 -> 不重试")
    else:
        return "成功"

# ========== 根据返回值判断 ==========
@retry(
    stop=stop_after_attempt(5),
    retry=retry_if_result(lambda x: x is None or x == "")
)
def retry_on_bad_result():
    """返回None或空字符串就重试"""
    import random
    result = random.choice([None, "", "有效数据"])
    print(f"返回: {result}")
    return result

# ========== 自定义重试逻辑 ==========
def should_retry(exception):
    """自定义判断函数"""
    error_msg = str(exception)
    # 包含这些关键词才重试
    retry_keywords = ['timeout', '503', 'rate limit', 'temporary']
    return any(keyword in error_msg.lower() for keyword in retry_keywords)

@retry(
    stop=stop_after_attempt(3),
    retry=retry_if_exception(should_retry)
)
def custom_retry_logic():
    """根据错误信息决定是否重试"""
    import random
    errors = [
        "Request timeout",           # 会重试
        "Error 503 Service Down",    # 会重试  
        "Invalid API key",           # 不重试
        "Rate limit exceeded"        # 会重试
    ]
    error = random.choice(errors)
    raise Exception(error)

if __name__ == '__main__':
    print("1. 选择性异常重试")
    print("-" * 40)
    for i in range(3):
        try:
            result = selective_retry()
            print(f"✅ {result}")
        except Exception as e:
            print(f"❌ {e}")
    
    print("\n2. 基于返回值重试")
    print("-" * 40)
    result = retry_on_bad_result()
    print(f"✅ 最终结果: {result}")
    
    print("\n3. 自定义重试条件")
    print("-" * 40)
    for i in range(3):
        try:
            custom_retry_logic()
        except Exception as e:
            print(f"❌ 不重试: {e}")

Demo 4: 实战模拟

"""
demo4_real_world.py - 模拟真实场景
"""
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_log, after_log
)
import logging
import time
import random

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)
logger = logging.getLogger(__name__)

# ========== 文件上传模拟 ==========
class FileUploader:
    def __init__(self):
        self.uploaded_files = []
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(ConnectionError),
        before=lambda _: logger.info("🔄 准备上传文件..."),
        after=lambda _: logger.info("📁 上传尝试结束")
    )
    def upload(self, filename):
        """模拟文件上传（70%成功率）"""
        logger.info(f"上传中: {filename}")
        
        # 模拟网络波动
        if random.random() < 0.7:
            raise ConnectionError(f"上传{filename}时网络断开")
        
        self.uploaded_files.append(filename)
        logger.info(f"✅ {filename} 上传成功")
        return True

# ========== 网页爬虫模拟 ==========
class WebScraper:
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=2, min=1, max=60),
        retry=retry_if_exception_type(TimeoutError)
    )
    def fetch_page(self, url):
        """模拟爬取网页（防反爬）"""
        logger.info(f"爬取: {url}")
        
        # 模拟各种失败
        fail_type = random.random()
        if fail_type < 0.5:
            raise TimeoutError("请求超时")
        elif fail_type < 0.8:
            raise ConnectionError("连接被拒绝")
        
        return f"<html>{url}的内容</html>"

# ========== 数据库操作模拟 ==========
class DatabaseOperator:
    def __init__(self):
        self.retry_count = 0
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=0.5, min=1, max=5)
    )
    def execute_with_retry(self, query):
        """模拟数据库操作"""
        self.retry_count += 1
        logger.info(f"执行SQL(第{self.retry_count}次): {query}")
        
        # 模拟死锁
        if random.random() < 0.6:
            raise Exception("Deadlock detected")
        
        self.retry_count = 0
        return "Query executed successfully"

if __name__ == '__main__':
    print("=" * 60)
    print("🔧 真实场景模拟")
    print("=" * 60)
    
    # 场景1: 文件上传
    print("\n📂 场景1: 批量文件上传")
    print("-" * 40)
    uploader = FileUploader()
    files = ["data1.txt", "data2.txt", "data3.txt"]
    
    for file in files:
        try:
            uploader.upload(file)
        except Exception as e:
            logger.error(f"❌ {file} 上传失败: {e}")
    
    logger.info(f"成功上传: {uploader.uploaded_files}")
    
    # 场景2: 网页爬取
    print("\n🌐 场景2: 网页爬取")
    print("-" * 40)
    scraper = WebScraper()
    
    try:
        content = scraper.fetch_page("https://example.com")
        logger.info(f"✅ 爬取成功: {content}")
    except Exception as e:
        logger.error(f"❌ 爬取失败: {e}")
    
    # 场景3: 数据库操作
    print("\n💾 场景3: 数据库重试")
    print("-" * 40)
    db = DatabaseOperator()
    
    try:
        result = db.execute_with_retry("SELECT * FROM users")
        logger.info(f"✅ {result}")
    except Exception as e:
        logger.error(f"❌ 数据库操作失败: {e}")

Demo 5: 回调函数和监控

"""
demo5_callbacks.py - 掌握重试过程
"""
from tenacity import (
    retry, stop_after_attempt, wait_fixed,
    before_sleep, after_attempt, before_log,
    RetryCallState
)
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ========== 监控每次重试 ==========
def on_retry(retry_state: RetryCallState):
    """重试前回调"""
    print(f"⚠️ 重试 #{retry_state.attempt_number}")
    print(f"   等待 {retry_state.next_action.sleep} 秒")
    
    # 访问异常信息
    if retry_state.outcome.failed:
        exception = retry_state.outcome.exception()
        print(f"   原因: {type(exception).__name__}: {exception}")

def on_success(retry_state: RetryCallState):
    """成功回调"""
    if retry_state.attempt_number > 1:
        print(f"🎉 在第 {retry_state.attempt_number} 次尝试后成功!")

@retry(
    stop=stop_after_attempt(5),
    wait=wait_fixed(1),
    before_sleep=on_retry,
    after=on_success
)
def monitored_function():
    """带监控的函数"""
    import random
    if random.random() < 0.8:
        raise Exception("随机失败")
    return "成功!"

# ========== 统计重试信息 ==========
class RetryStats:
    def __init__(self):
        self.total_retries = 0
        self.total_time = 0
        self.start_time = None
    
    def before_retry(self, retry_state):
        if self.start_time is None:
            self.start_time = time.time()
        self.total_retries += 1
    
    def get_stats(self):
        elapsed = time.time() - self.start_time if self.start_time else 0
        return {
            'total_retries': self.total_retries,
            'total_time': f"{elapsed:.2f}s"
        }

stats = RetryStats()

@retry(
    stop=stop_after_attempt(4),
    wait=wait_fixed(0.5),
    before_sleep=stats.before_retry
)
def stats_function():
    """收集统计信息"""
    import random
    if random.random() < 0.7:
        raise Exception("失败")
    return "成功"

# ========== 条件打断 ==========
def should_abort(retry_state):
    """根据时间决定是否继续"""
    elapsed = time.time() - retry_state.start_time
    return elapsed > 10  # 超过10秒就放弃

@retry(
    stop=stop_after_attempt(100),  # 理论上100次
    wait=wait_fixed(1),
    retry=lambda e: True  # 任何异常都重试
)
def abortable_function():
    """可能被打断的函数"""
    elapsed = time.time() - abortable_function.start_time
    if elapsed > 5:
        print(f"⏰ 已过 {elapsed:.1f}秒，主动放弃")
        raise Exception("超时放弃")
    
    print(f"⏳ 执行中... ({elapsed:.1f}秒)")
    raise Exception("继续重试")

abortable_function.start_time = time.time()

if __name__ == '__main__':
    print("=" * 50)
    print("1. 监控重试过程")
    print("=" * 50)
    try:
        result = monitored_function()
        print(f"✅ 结果: {result}")
    except Exception as e:
        print(f"❌ 最终失败: {e}")
    
    print("\n" + "=" * 50)
    print("2. 重试统计")
    print("=" * 50)
    try:
        stats_function()
    except:
        pass
    print(f"📊 统计: {stats.get_stats()}")
    
    print("\n" + "=" * 50)
    print("3. 超时放弃")
    print("=" * 50)
    try:
        abortable_function()
    except Exception as e:
        print(f"❌ {e}")

综合示例：

"""
demo_comprehensive.py - 综合实战：完整的重试监控系统
"""
import time
import random
import traceback
from datetime import datetime
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep,
    RetryCallState,
    RetryError
)


# ========== 工具函数 ==========

def format_time() -> str:
    """格式化当前时间"""
    return datetime.now().strftime("%H:%M:%S")


def get_error_detail(exception: Exception) -> str:
    """提取错误详情"""
    error_type = type(exception).__name__
    error_msg = str(exception)

    # 提取traceback的最后一行（关键错误位置）
    tb_lines = traceback.format_exc().split('\n')
    error_location = "未知位置"
    for line in reversed(tb_lines):
        if line.strip().startswith('File'):
            error_location = line.strip()
            break

    return error_type, error_msg, error_location


# ========== 自定义回调函数 ==========

def log_retry_attempt(retry_state: RetryCallState) -> None:
    """
    每次重试前打印详细信息
    这个函数会在每次重试之前自动调用
    """
    # 获取当前重试次数
    attempt_number = retry_state.attempt_number

    # 获取异常信息
    if retry_state.outcome and retry_state.outcome.failed:
        exception = retry_state.outcome.exception()
        error_type, error_msg, error_location = get_error_detail(exception)
    else:
        error_type = "首次尝试"
        error_msg = "无错误"
        error_location = "N/A"

    # 获取等待时间
    wait_seconds = retry_state.next_action.sleep if retry_state.next_action else 0

    # 打印分隔线
    print(f"\n{'=' * 60}")
    print(f"🔄 重试 #{attempt_number}")
    print(f"{'=' * 60}")
    print(f"⏰ 时间: {format_time()}")
    print(f"💥 错误类型: {error_type}")
    print(f"📝 错误信息: {error_msg}")
    print(f"📍 错误位置: {error_location}")
    print(f"⏳ 等待时间: {wait_seconds:.1f} 秒")
    print(f"{'=' * 60}\n")


def log_after_attempt(retry_state: RetryCallState) -> None:
    """每次尝试后打印"""
    if retry_state.outcome and retry_state.outcome.failed:
        # 失败
        pass  # 失败信息已在 before_sleep 中打印
    else:
        # 成功
        total_attempts = retry_state.attempt_number
        if total_attempts > 1:
            print(f"🎉 在第 {total_attempts} 次尝试后终于成功！")
        else:
            print(f"✅ 一次就成功了！")


# ========== 场景模拟 ==========

class NetworkSimulator:
    """模拟各种网络/业务异常"""

    @staticmethod
    def raise_random_error():
        """随机抛出不同类型的异常"""
        error_type = random.random()

        if error_type < 0.3:
            # 30% 概率超时
            raise TimeoutError("请求超时：服务器响应时间超过30秒")
        elif error_type < 0.5:
            # 20% 概率连接错误
            raise ConnectionError("连接被拒绝：目标服务器拒绝连接")
        elif error_type < 0.7:
            # 20% 概率临时性服务器错误
            raise Exception("服务器返回 503 Service Temporarily Unavailable")
        elif error_type < 0.85:
            # 15% 概率限流
            raise Exception("API限流：请求频率过高，请稍后重试")
        else:
            # 15% 概率其他错误
            raise RuntimeError("未知运行时错误：数据解析失败")


# ========== 实战示例1：API调用 ==========

class APIClient:
    """模拟API客户端"""

    def __init__(self, api_name: str):
        self.api_name = api_name
        self.call_count = 0

    @retry(
        stop=stop_after_attempt(5),  # 最多重试5次
        wait=wait_exponential(multiplier=1, min=2, max=30),  # 指数退避
        retry=retry_if_exception_type((  # 只重试这些异常
                TimeoutError,
                ConnectionError,
                Exception
        )),
        before_sleep=log_retry_attempt,  # 重试前打印
        after=log_after_attempt  # 每次尝试后处理
    )
    def call_api(self, endpoint: str, data: dict = None) -> dict:
        """
        调用API接口
        会自动重试：超时、连接错误、503、限流等
        """
        self.call_count += 1
        print(f"🌐 [{self.api_name}] 第{self.call_count}次调用 API: {endpoint}")

        # 模拟网络延迟
        time.sleep(0.5)

        # 70% 概率失败
        if random.random() < 0.7:
            NetworkSimulator.raise_random_error()

        # 成功返回
        self.call_count = 0
        return {
            "status": "success",
            "data": f"{endpoint} 的响应数据",
            "timestamp": format_time()
        }


# ========== 实战示例2：文件处理 ==========

class FileProcessor:
    """模拟文件处理器"""

    def __init__(self):
        self.processed_count = 0

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=3, max=15),
        retry=retry_if_exception_type((
                TimeoutError,
                ConnectionError,
                OSError,
                IOError
        )),
        before_sleep=log_retry_attempt,
        after=log_after_attempt
    )
    def process_file(self, filename: str) -> dict:
        """
        处理文件
        会自动重试：超时、IO错误等
        """
        print(f"📄 开始处理文件: {filename}")

        # 模拟文件操作
        time.sleep(1)

        # 模拟各种失败情况
        fail_probability = random.random()
        if fail_probability < 0.4:
            raise TimeoutError(f"处理超时：文件 {filename} 过大，处理超过60秒")
        elif fail_probability < 0.7:
            raise IOError(f"IO错误：无法读取文件 {filename}，磁盘繁忙")
        elif fail_probability < 0.85:
            raise ConnectionError(f"连接中断：云存储服务临时不可用")

        # 成功
        self.processed_count += 1
        return {
            "filename": filename,
            "status": "processed",
            "size": random.randint(1000, 50000)
        }


# ========== 实战示例3：数据库操作 ==========

class DatabaseClient:
    """模拟数据库客户端"""

    def __init__(self):
        self.query_count = 0
        self.connection_pool = 10

    @retry(
        stop=stop_after_attempt(4),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type(Exception),
        before_sleep=log_retry_attempt,
        after=log_after_attempt
    )
    def execute_query(self, sql: str) -> list:
        """
        执行数据库查询
        会自动重试：死锁、连接池耗尽等
        """
        self.query_count += 1
        print(f"💾 [查询#{self.query_count}] 执行: {sql[:50]}...")

        time.sleep(0.3)

        error_random = random.random()
        if error_random < 0.3:
            raise Exception("死锁检测：事务被选为死锁牺牲品，请重试")
        elif error_random < 0.6:
            raise Exception("连接池耗尽：所有连接都在使用中")
        elif error_random < 0.8:
            raise Exception("查询超时：statement_timeout 超过限制")

        # 成功
        self.query_count = 0
        return [
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"}
        ]


# ========== 执行演示 ==========

def run_demo():
    """运行完整演示"""
    print("=" * 70)
    print("🚀 Tenacity 综合重试演示系统")
    print("=" * 70)
    print("功能：自动重试 + 详细日志 + 错误追踪")
    print("=" * 70)

    # ---------- 场景1 ----------
    print("\n" + "=" * 35)
    print("场景1: API 调用自动重试")
    print("=" * 35)

    api_client = APIClient("千问API")

    try:
        result = api_client.call_api("/v1/chat", {"prompt": "你好"})
        print(f"\n✅ API调用成功!")
        print(f"返回数据: {result}")
    except RetryError as e:
        print(f"\n❌ API调用最终失败!")
        print(f"总共尝试了 {e.last_attempt.attempt_number} 次")
        print(f"最后一次错误: {e.last_attempt.exception()}")

    time.sleep(2)

    # ---------- 场景2 ----------
    print("\n" + "=" * 35)
    print("场景2: 文件处理自动重试")
    print("=" * 35)

    file_processor = FileProcessor()
    files = ["data_batch_1.txt", "data_batch_2.txt", "data_batch_3.txt"]

    for filename in files:
        try:
            result = file_processor.process_file(filename)
            print(f"✅ {filename} 处理成功: 大小={result['size']}字节")
        except RetryError as e:
            print(f"❌ {filename} 处理失败!")
            original_error = e.last_attempt.exception()
            error_type = type(original_error).__name__
            print(f"   错误类型: {error_type}")
            print(f"   错误信息: {original_error}")
        print("-" * 50)

    time.sleep(2)

    # ---------- 场景3 ----------
    print("\n" + "=" * 35)
    print("场景3: 数据库操作自动重试")
    print("=" * 35)

    db_client = DatabaseClient()
    queries = [
        "SELECT * FROM users WHERE active = 1",
        "UPDATE orders SET status = 'done'",
        "INSERT INTO logs VALUES (...)"
    ]

    for query in queries:
        try:
            results = db_client.execute_query(query)
            print(f"✅ 查询成功，返回 {len(results)} 条记录")
        except RetryError as e:
            print(f"❌ 查询最终失败!")
            print(f"   SQL: {query[:30]}...")
            print(f"   最后一次错误: {e.last_attempt.exception()}")
        print("-" * 50)


if __name__ == '__main__':
    run_demo()
下一篇
tenacity 重试监控实战
上一篇 drssionpage实战qwen自动化