Python利用tenacity库处理超时重试机制详解

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def test_function():
    print("尝试执行...")
    raise Exception("发生错误")

if __name__ == '__main__':
    try:
        test_function()
    except Exception as e:
        print(f"最终失败: {e}")

import requests
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(5), wait=wait_fixed(3))
def fetch_data(url):
    response = requests.get(url)
    response.raise_for_status()  # 如果响应状态码不是 200，将抛出异常
    return response.json()

url = "https://api.example.com/data"
try:
    data = fetch_data(url)
    print("数据获取成功:", data)
except Exception as e:
    print(f"数据获取失败: {e}")

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def read_file(file_path):
    with open(file_path, 'r') as file:
        return file.read()

try:
    content = read_file("example.txt")
    print("文件内容:", content)
except Exception as e:
    print(f"读取文件失败: {e}")

import sqlite3
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def execute_query(query):
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    cursor.execute(query)
    conn.commit()
    conn.close()

try:
    execute_query("INSERT INTO users (name) VALUES ('Alice')")
    print("插入成功")
except Exception as e:
    print(f"数据库操作失败: {e}")