Python容错安全案例如何保障异常容错

wen python案例 2026-06-28 253

本文目录导读：

基础异常处理模式
重试机制（Retry Pattern）
上下文管理器（Context Manager）
优雅降级（Graceful Degradation）
日志和监控
防御性编程
断言和契约编程
综合示例：安全的Web API处理

Python中的容错安全主要通过异常处理机制、上下文管理器、断言和日志记录等工具来实现，下面通过几个典型的安全案例，详细说明如何保障异常容错。

基础异常处理模式

案例：文件读取容错

def read_config_file(filepath):
    """
    安全读取配置文件，处理各种可能的异常
    """
    try:
        with open(filepath, 'r') as file:
            content = file.read()
            return content
    except FileNotFoundError:
        # 文件不存在时使用默认配置
        print(f"配置文件 {filepath} 未找到，使用默认配置")
        return DEFAULT_CONFIG
    except PermissionError:
        # 权限不足时返回空配置
        print(f"无权限读取文件 {filepath}")
        return {}
    except UnicodeDecodeError:
        # 编码错误时尝试其他编码
        try:
            with open(filepath, 'r', encoding='utf-8-sig') as file:
                return file.read()
        except:
            return {}
    except Exception as e:
        # 记录未知异常，避免程序崩溃
        log_error(f"读取配置文件异常: {e}")
        return {}

重试机制（Retry Pattern）

案例：网络请求重试

import time
from functools import wraps
def retry(max_attempts=3, delay=1, backoff=2, exceptions=(Exception,)):
    """
    重试装饰器，用于网络请求等不稳定的操作
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            current_delay = delay
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    attempts += 1
                    if attempts == max_attempts:
                        raise
                    print(f"第{attempts}次尝试失败: {e}，{current_delay}秒后重试...")
                    time.sleep(current_delay)
                    current_delay *= backoff  # 指数退避
            return None
        return wrapper
    return decorator
@retry(max_attempts=3, delay=1, exceptions=(ConnectionError, TimeoutError))
def fetch_data_from_api(url):
    """从API获取数据，自动重试"""
    # 模拟网络请求
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

上下文管理器（Context Manager）

案例：数据库连接安全

class DatabaseConnection:
    """安全的数据库连接管理"""
    def __init__(self, host, port, database):
        self.host = host
        self.port = port
        self.database = database
        self.connection = None
    def __enter__(self):
        # 建立连接
        try:
            self.connection = create_connection(self.host, self.port, self.database)
            return self.connection
        except ConnectionError as e:
            raise ConnectionError(f"数据库连接失败: {e}")
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 确保连接被关闭
        if self.connection:
            try:
                self.connection.close()
            except Exception as e:
                print(f"关闭数据库连接时发生错误: {e}")
        # 处理异常
        if exc_type is not None:
            print(f"数据库操作异常: {exc_type.__name__}: {exc_val}")
            # 返回False表示不抑制异常，True表示抑制
            return False
# 使用示例
def query_user_data(user_id):
    """安全查询用户数据"""
    with DatabaseConnection('localhost', 5432, 'users') as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        return cursor.fetchone()

优雅降级（Graceful Degradation）

案例：功能降级

class PaymentProcessor:
    """支付处理器，支持优雅降级"""
    def __init__(self):
        self.primary_gateway = PrimaryPaymentGateway()
        self.backup_gateway = BackupPaymentGateway()
    def process_payment(self, amount, card_info):
        """
        处理支付，主网关失败时自动降级到备份网关
        """
        # 尝试主网关
        try:
            result = self.primary_gateway.charge(amount, card_info)
            return {"status": "success", "gateway": "primary", "result": result}
        except GatewayTimeoutError:
            print("主网关超时，尝试备份网关")
        except GatewayUnavailableError:
            print("主网关不可用，尝试备份网关")
        except PaymentValidationError as e:
            # 参数验证错误，不需要降级
            return {"status": "failed", "error": str(e)}
        # 尝试备份网关
        try:
            result = self.backup_gateway.charge(amount, card_info)
            return {"status": "success", "gateway": "backup", "result": result}
        except Exception as e:
            print(f"所有支付网关均失败: {e}")
            return {"status": "failed", "error": "所有支付通道不可用"}

日志和监控

案例：完整的异常记录

import logging
import sys
from datetime import datetime
class ErrorLogger:
    """异常日志记录器"""
    def __init__(self, log_file='error.log'):
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.ERROR)
        # 文件处理器
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.ERROR)
        # 控制台处理器
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.ERROR)
        # 格式化器
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
    def log_exception(self, func_name, exception, context=None):
        """记录异常信息"""
        error_info = {
            'function': func_name,
            'exception_type': type(exception).__name__,
            'exception_message': str(exception),
            'timestamp': datetime.now().isoformat(),
            'context': context or {}
        }
        self.logger.error(
            f"异常发生在 {func_name}: {exception}",
            extra={'error_info': error_info}
        )
# 使用示例
error_logger = ErrorLogger()
def critical_operation(data):
    """关键操作，需要完整记录异常"""
    try:
        result = process_data(data)
        return result
    except Exception as e:
        error_logger.log_exception(
            'critical_operation', 
            e, 
            {'data_size': len(data), 'data_type': type(data).__name__}
        )
        return None

防御性编程

案例：输入验证和安全默认值

def safe_int_conversion(value, default=0):
    """安全的整数转换"""
    if value is None:
        return default
    try:
        result = int(value)
        return result
    except (ValueError, TypeError):
        return default
    except OverflowError:
        return default
def process_user_input(input_data):
    """
    处理用户输入，使用安全默认值
    """
    # 使用安全获取方法
    name = input_data.get('name', '匿名用户')[:50]  # 限制长度
    age = safe_int_conversion(input_data.get('age'), 0)
    email = input_data.get('email', '')
    # 验证邮箱格式
    if '@' not in email:
        email = ''
    return {
        'name': name,
        'age': max(0, min(age, 150)),  # 限制年龄范围
        'email': email
    }

断言和契约编程

class BankAccount:
    """银行账户，使用断言确保状态一致性"""
    def __init__(self, account_number, initial_balance=0):
        assert account_number and len(account_number) == 10, "无效的账号"
        assert initial_balance >= 0, "初始余额不能为负"
        self.account_number = account_number
        self.balance = initial_balance
        self.is_active = True
    def withdraw(self, amount):
        """
        取款操作，包含前置条件和后置条件检查
        """
        # 前置条件
        assert amount > 0, "取款金额必须大于0"
        assert self.is_active, "账户已停用"
        assert self.balance >= amount, "余额不足"
        old_balance = self.balance
        try:
            self.balance -= amount
            transaction_id = generate_transaction_id()
            # 后置条件
            assert self.balance == old_balance - amount, "余额计算错误"
            assert self.balance >= 0, "余额不能为负"
            return {"success": True, "transaction_id": transaction_id}
        except Exception as e:
            # 发生异常时回滚
            self.balance = old_balance
            raise

综合示例：安全的Web API处理

class SafeAPIHandler:
    """安全的API处理器"""
    def handle_request(self, request_data):
        """
        处理API请求，包含多层容错
        """
        try:
            # 1. 数据验证
            validated_data = self._validate_request(request_data)
            if not validated_data:
                return {"error": "数据验证失败"}, 400
            # 2. 权限检查
            if not self._check_permission(validated_data):
                return {"error": "无权限"}, 403
            # 3. 执行业务逻辑
            result = self._process_data(validated_data)
            # 4. 缓存结果
            self._cache_result(result)
            return {"data": result}, 200
        except RateLimitExceeded as e:
            # 限流异常
            return {"error": "请求频率过高"}, 429
        except ServiceUnavailableError as e:
            # 服务不可用，返回友好的错误信息
            return {"error": "服务暂时不可用，请稍后重试"}, 503
        except TimeoutError as e:
            # 超时处理
            log_timeout_error(e)
            return {"error": "请求超时"}, 504
        except Exception as e:
            # 未知异常，记录但不暴露细节
            log_unexpected_error(e)
            return {"error": "处理请求时发生错误"}, 500
    def _validate_request(self, data):
        """数据验证，返回验证后的数据或None"""
        try:
            # 使用try-except包装整个验证过程
            schema = RequestSchema()
            return schema.load(data)
        except ValidationError as e:
            print(f"数据验证失败: {e.messages}")
            return None
        except Exception as e:
            print(f"验证过程中发生未知错误: {e}")
            return None