Python装饰器之重试机制，处理临时故障！

def retry_on_failure(retries=3):

def decorator(func):

def wrapper(*args, **kwargs):

attempt = 0

while attempt <= retries:

try:

return func(*args, **kwargs)

except Exception:

attempt += 1

if attempt > retries:

raise

return wrapper

return decorator

@retry_on_failure()

def might_fail_randomly():

import random

if random.randint(0, 9) > 5:  # 模拟随机失败

raise Exception("Temporary failure.")

return "Success!"

print(might_fail_randomly())

import time

def retry_with_delay(retries=3, delay=1):

def decorator(func):

def wrapper(*args, **kwargs):

attempt = 0

while attempt <= retries:

try:

return func(*args, **kwargs)

except Exception:

attempt += 1

if attempt <= retries:

time.sleep(delay)

else:

raise

return wrapper

return decorator

@retry_with_delay(retries=5, delay=2)

def simulate_network_issue():

import random

if random.randint(0, 9) > 7:  # 模拟网络不稳定

raise IOError("Network issue.")

return "Data fetched."

print(simulate_network_issue())

def retry_on_specific_errors(retries=3, delay=1, error_types=(Exception,)):

def decorator(func):

def wrapper(*args, **kwargs):

attempt = 0

while attempt <= retries:

try:

return func(*args, **kwargs)

except tuple(error_types) as e:

attempt += 1

if attempt <= retries:

time.sleep(delay)

print(f"Retrying due to {e.__class__.__name__}.")

else:

raise

return wrapper

return decorator

@retry_on_specific_errors(retries=3, error_types=(ConnectionError,))

def fetch_remote_data(url):

# 这里假设了fetch_data函数会抛出ConnectionError

raise ConnectionError("Failed to connect.")

try:

fetch_remote_data("http://example.com/data")

except Exception as e:

print(f"Failed after retries: {e}")