Python容错工具案例如何封装异常容错

本文目录导读：

1. 装饰器方式封装
2. 上下文管理器封装
3. 类封装方式
4. 异步任务容错
5. 配置文件驱动的容错
6. 最佳实践建议

我来分享几个Python异常容错封装的实用案例：

装饰器方式封装

基础重试装饰器

import time
import functools
from typing import Callable, Tuple, Type
def retry(max_retries: int = 3, 
          delay: float = 1, 
          exceptions: Tuple[Type[Exception], ...] = (Exception,)):
    """
    异常重试装饰器
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        raise
                    print(f"第{attempt+1}次失败: {e}")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
# 使用示例
@retry(max_retries=3, delay=0.5, exceptions=(ValueError, ConnectionError))
def fetch_data(url: str) -> dict:
    # 模拟异常
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络连接失败")
    return {"data": "成功获取的数据"}
# 测试
result = fetch_data("http://example.com")
print(result)

兜底值装饰器

import functools
def fallback(default_value=None):
    """
    异常时返回默认值
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                print(f"执行异常: {e}，返回默认值")
                return default_value
        return wrapper
    return decorator
# 使用示例
@fallback(default_value="未知用户")
def get_username(user_id: int) -> str:
    # 模拟可能异常的操作
    if user_id <= 0:
        raise ValueError("用户ID无效")
    return f"用户{user_id}"
print(get_username(1))     # 正常
print(get_username(-1))    # 返回默认值

上下文管理器封装

超时容错上下文

import signal
import time
from contextlib import contextmanager
class TimeoutError(Exception):
    pass
@contextmanager
def timeout(seconds: int):
    """
    超时容错上下文管理器
    """
    def handler(signum, frame):
        raise TimeoutError("操作超时")
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)
    try:
        yield
    except TimeoutError as e:
        print(f"超时容错: {e}")
        # 这里可以进行补偿操作
    except Exception as e:
        print(f"其他异常容错: {e}")
    finally:
        signal.alarm(0)  # 取消闹钟
# 使用示例
with timeout(seconds=2):
    time.sleep(1)  # 正常执行
with timeout(seconds=2):
    time.sleep(3)  # 触发超时容错

资源清理上下文

from contextlib import contextmanager
import sqlite3
@contextmanager
def safe_db_connection(db_path: str):
    """
    数据库安全连接上下文管理器
    """
    conn = None
    try:
        conn = sqlite3.connect(db_path)
        yield conn
    except sqlite3.Error as e:
        print(f"数据库操作异常: {e}")
        if conn:
            conn.rollback()  # 回滚事务
    except Exception as e:
        print(f"其他异常: {e}")
        if conn:
            conn.rollback()
    finally:
        if conn:
            conn.close()
            print("数据库连接已关闭")
# 使用示例
with safe_db_connection("test.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM non_existent_table")

类封装方式

通用容错处理类

import functools
import logging
from typing import Callable, Any
class FaultTolerance:
    """
    通用容错处理类
    """
    def __init__(self, 
                 max_retries: int = 3,
                 fallback_value: Any = None,
                 logger: logging.Logger = None):
        self.max_retries = max_retries
        self.fallback_value = fallback_value
        self.logger = logger or logging.getLogger(__name__)
    def retry(self, func: Callable, *args, **kwargs):
        """
        重试机制
        """
        last_exception = None
        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                last_exception = e
                self.logger.warning(
                    f"第{attempt+1}次执行失败: {str(e)}"
                )
                if attempt < self.max_retries - 1:
                    import time
                    time.sleep(0.5 * (attempt + 1))
        self.logger.error(
            f"重试{self.max_retries}次后仍然失败"
        )
        return self.fallback_value
    def safe_execute(self, func: Callable, *args, **kwargs):
        """
        安全执行，捕获所有异常
        """
        try:
            return func(*args, **kwargs)
        except KeyboardInterrupt:
            raise  # 不捕获键盘中断
        except Exception as e:
            self.logger.error(f"执行异常: {str(e)}")
            return self.fallback_value
    def __call__(self, func: Callable) -> Callable:
        """
        作为装饰器使用
        """
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return self.retry(func, *args, **kwargs)
        return wrapper
# 配置日志
logging.basicConfig(level=logging.INFO)
# 使用示例
ft = FaultTolerance(max_retries=2, fallback_value=None)
@ft
def unstable_api_call(user_id: int) -> dict:
    import random
    if random.random() < 0.6:
        raise ConnectionError("API不可用")
    return {"user_id": user_id, "name": "Test"}
# 批量处理多个任务
tasks = [1, 2, 3, 4, 5]
results = []
for task in tasks:
    result = ft.safe_execute(unstable_api_call, task)
    results.append(result)
print(f"处理结果: {results}")

异步任务容错

import asyncio
import functools
from typing import Callable, Any
class AsyncFaultTolerance:
    """
    异步任务容错处理
    """
    def __init__(self, 
                 max_retries: int = 3,
                 circuit_breaker_threshold: int = 5):
        self.max_retries = max_retries
        self.circuit_breaker_threshold = circuit_breaker_threshold
        self.failure_count = 0
        self.circuit_open = False
    async def retry_async(self, func: Callable, *args, **kwargs):
        """
        异步重试机制
        """
        if self.circuit_open:
            print("熔断器已打开，拒绝请求")
            return None
        for attempt in range(self.max_retries):
            try:
                if asyncio.iscoroutinefunction(func):
                    result = await func(*args, **kwargs)
                else:
                    result = func(*args, **kwargs)
                # 成功后重置故障计数
                self.failure_count = 0
                return result
            except Exception as e:
                self.failure_count += 1
                print(f"第{attempt+1}次失败: {e}")
                # 检查熔断器
                if self.failure_count >= self.circuit_breaker_threshold:
                    self.circuit_open = True
                    print("熔断器触发!")
                    break
                await asyncio.sleep(0.5 * (attempt + 1))
        return None
# 使用示例
async def unstable_async_api(url: str) -> dict:
    import random
    await asyncio.sleep(0.1)
    if random.random() < 0.7:
        raise ConnectionError(f"连接 {url} 失败")
    return {"url": url, "status": "success"}
async def main():
    ft = AsyncFaultTolerance(max_retries=3)
    # 测试多个异步请求
    urls = [f"http://api.example.com/{i}" for i in range(3)]
    tasks = [ft.retry_async(unstable_async_api, url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(f"异步处理结果: {results}")
# 运行
asyncio.run(main())

配置文件驱动的容错

import json
from typing import Dict, Any
class ConfigDrivenFT:
    """
    配置驱动的容错策略
    """
    def __init__(self, config_path: str = None):
        self.config = {
            "retry": {
                "enabled": True,
                "max_retries": 3,
                "delay": 1,
                "backoff_factor": 2
            },
            "circuit_breaker": {
                "enabled": True,
                "failure_threshold": 5,
                "recovery_timeout": 30
            },
            "fallback": {
                "enabled": True,
                "default_value": None
            }
        }
        if config_path:
            self.load_config(config_path)
    def load_config(self, config_path: str):
        """加载配置文件"""
        try:
            with open(config_path, 'r') as f:
                user_config = json.load(f)
                self.config.update(user_config)
        except FileNotFoundError:
            print(f"配置文件 {config_path} 不存在，使用默认配置")
        except json.JSONDecodeError:
            print("配置文件格式错误，使用默认配置")
    def execute_with_ft(self, func, *args, **kwargs):
        """根据配置执行容错策略"""
        if self.config["retry"]["enabled"]:
            return self._retry_strategy(func, *args, **kwargs)
        else:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if self.config["fallback"]["enabled"]:
                    return self.config["fallback"]["default_value"]
                raise
    def _retry_strategy(self, func, *args, **kwargs):
        """重试策略实现"""
        max_retries = self.config["retry"]["max_retries"]
        delay = self.config["retry"]["delay"]
        backoff = self.config["retry"]["backoff_factor"]
        import time
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    if self.config["fallback"]["enabled"]:
                        return self.config["fallback"]["default_value"]
                    raise
                wait_time = delay * (backoff ** attempt)
                print(f"重试第{attempt+1}次，等待{wait_time}秒")
                time.sleep(wait_time)
        return self.config["fallback"]["default_value"]
# 配置文件示例 fault_tolerance.json
"""
{
    "retry": {
        "enabled": true,
        "max_retries": 5,
        "delay": 0.5,
        "backoff_factor": 1.5
    },
    "fallback": {
        "enabled": true,
        "default_value": "fallback_data"
    }
}
"""
# 使用示例
ft = ConfigDrivenFT("fault_tolerance.json")
def unstable_function(data: str) -> str:
    import random
    if random.random() < 0.6:
        raise ValueError("模拟异常")
    return f"处理: {data}"
result = ft.execute_with_ft(unstable_function, "测试数据")
print(f"容错处理结果: {result}")

最佳实践建议

1. 分层容错：在不同层面（函数、模块、服务）实施合适的容错策略
2. 日志记录：详细记录异常信息，便于调试和监控
3. 适当重试：设置合理的重试次数和延迟策略
4. 降级处理：提供备选方案或默认值
5. 熔断机制：防止雪崩效应
6. 监控告警：异常次数过多时及时告警

这些案例可以根据实际需求组合使用,构建健壮的容错系统。