Python异常处理完全指南：从基础try-except到自定义异常体系

一、基础异常处理：从入门到精通

1. 基本try-except语法与常见陷阱

基础语法：

# 基础异常捕获
try:
    # 可能引发异常的代码
    result = 10 / 0
    print(f"结果是: {result}")
except ZeroDivisionError:
    # 处理特定异常
    print("错误：除数不能为零！")

# 捕获多个异常
try:
    number = int("abc")  # ValueError
    data = [1, 2, 3]
    print(data[10])      # IndexError
except (ValueError, IndexError) as e:
    print(f"数据错误: {e}")

# 分别处理不同异常
try:
    file = open("nonexistent.txt", "r")
    content = file.read()
    number = int(content)
except FileNotFoundError:
    print("文件不存在")
except ValueError:
    print("文件内容不是有效数字")
except Exception as e:
    print(f"其他错误: {e}")
finally:
    # 无论是否发生异常都会执行
    print("清理操作完成")

常见陷阱与解决方案：

# 陷阱1：过于宽泛的异常捕获
# 反模式
try:
    do_something()
except:  # 捕获所有异常，包括KeyboardInterrupt等系统异常
    print("发生错误")

# 正确做法：明确指定异常类型
try:
    do_something()
except (ValueError, TypeError) as e:
    print(f"业务逻辑错误: {e}")
except Exception as e:  # 捕获其他非系统异常
    print(f"意外错误: {e}")
    # 记录日志
    import logging
    logging.exception("操作失败")

# 陷阱2：忽略异常信息
# 反模式
try:
    risky_operation()
except Exception:
    pass  # 静默忽略，难以调试

# 正确做法：记录异常信息
try:
    risky_operation()
except Exception as e:
    print(f"操作失败: {e}")
    # 或者记录到日志
    logging.error(f"操作失败: {e}", exc_info=True)

# 陷阱3：不完整的清理
# 反模式
try:
    file = open("data.txt", "w")
    file.write(some_data)
    # 如果这里发生异常，文件不会关闭
    process_data()
except Exception as e:
    print(f"错误: {e}")

# 正确做法：使用with语句或finally
# 方案1：使用with语句（推荐）
try:
    with open("data.txt", "w") as file:
        file.write(some_data)
        process_data()  # 即使这里异常，文件也会正确关闭
except Exception as e:
    print(f"错误: {e}")

# 方案2：使用finally确保清理
file = None
try:
    file = open("data.txt", "w")
    file.write(some_data)
    process_data()
except Exception as e:
    print(f"错误: {e}")
finally:
    if file:
        file.close()

2. 异常对象与异常信息获取

def detailed_exception_handling():
    """详细的异常信息处理"""
    try:
        # 模拟多种可能异常的场景
        data = {"users": ["Alice", "Bob"]}
        
        # 场景1：键错误
        # print(data["nonexistent_key"])
        
        # 场景2：类型错误
        # result = "hello" + 123
        
        # 场景3：索引错误
        print(data["users"][5])
        
    except KeyError as e:
        print(f"键错误: 缺少键 {e}")
        print(f"可用键: {list(data.keys())}")
        
    except TypeError as e:
        print(f"类型错误: {e}")
        print("请检查数据类型是否匹配")
        
    except IndexError as e:
        print(f"索引错误: {e}")
        print(f"列表长度: {len(data['users'])}")
        
    except Exception as e:
        # 获取异常的详细信息
        print(f"异常类型: {type(e).__name__}")
        print(f"异常信息: {e}")
        print(f"异常参数: {e.args}")
        
        # 获取调用栈信息（调试用）
        import traceback
        print("调用栈:")
        traceback.print_exc()

# 测试异常处理
detailed_exception_handling()

# 异常对象的其他有用属性
try:
    risky_call()
except Exception as e:
    # 获取异常的字符串表示
    error_str = str(e)
    
    # 获取异常的类名
    exception_name = e.__class__.__name__
    
    # 使用traceback模块获取详细信息
    import traceback
    full_traceback = traceback.format_exc()
    
    print(f"错误: {error_str}")
    print(f"异常类型: {exception_name}")
    print("完整跟踪:")
    print(full_traceback)

二、自定义异常体系设计

1. 基础自定义异常

class AppError(Exception):
    """应用程序基础异常"""
    
    def __init__(self, message, error_code=None, details=None):
        self.message = message
        self.error_code = error_code
        self.details = details or {}
        super().__init__(self.message)
    
    def __str__(self):
        if self.error_code:
            return f"[{self.error_code}] {self.message}"
        return self.message
    
    def to_dict(self):
        """转换为字典，用于API响应等场景"""
        return {
            "error": self.message,
            "code": self.error_code,
            "details": self.details
        }

# 具体的业务异常
class ValidationError(AppError):
    """数据验证错误"""
    
    def __init__(self, message, field=None, value=None):
        details = {}
        if field:
            details["field"] = field
        if value:
            details["value"] = value
        
        super().__init__(
            message=message,
            error_code="VALIDATION_ERROR",
            details=details
        )

class DatabaseError(AppError):
    """数据库操作错误"""
    
    def __init__(self, message, operation=None, table=None):
        details = {}
        if operation:
            details["operation"] = operation
        if table:
            details["table"] = table
        
        super().__init__(
            message=message,
            error_code="DATABASE_ERROR",
            details=details
        )

class NetworkError(AppError):
    """网络通信错误"""
    
    def __init__(self, message, url=None, status_code=None):
        details = {}
        if url:
            details["url"] = url
        if status_code:
            details["status_code"] = status_code
        
        super().__init__(
            message=message,
            error_code="NETWORK_ERROR",
            details=details
        )

# 使用自定义异常
def validate_user_age(age):
    if not isinstance(age, int):
        raise ValidationError(
            "年龄必须是整数",
            field="age",
            value=age
        )
    if age < 0:
        raise ValidationError(
            "年龄不能为负数",
            field="age", 
            value=age
        )
    if age > 150:
        raise ValidationError(
            "年龄无效",
            field="age",
            value=age
        )
    return True

def create_user_profile(user_data):
    """创建用户档案"""
    try:
        # 验证数据
        validate_user_age(user_data.get("age"))
        
        # 模拟数据库操作
        if user_data.get("name") == "admin":
            raise DatabaseError(
                "用户名已存在",
                operation="INSERT",
                table="users"
            )
            
        print("用户创建成功")
        return {"status": "success", "user_id": 123}
        
    except ValidationError as e:
        print(f"数据验证失败: {e}")
        return e.to_dict()
    except DatabaseError as e:
        print(f"数据库操作失败: {e}")
        return e.to_dict()

# 测试自定义异常
test_cases = [
    {"name": "Alice", "age": 25},      # 成功
    {"name": "Bob", "age": -5},        # 验证错误
    {"name": "admin", "age": 30},      # 数据库错误
    {"name": "Charlie", "age": "old"}, # 类型错误
]

for user_data in test_cases:
    print(f"\n处理用户: {user_data}")
    result = create_user_profile(user_data)
    print(f"结果: {result}")

2. 层次化异常体系

class PaymentSystemError(AppError):
    """支付系统基础异常"""
    pass

class PaymentValidationError(PaymentSystemError):
    """支付验证错误"""
    
    def __init__(self, message, field=None, payment_method=None):
        details = {"payment_method": payment_method}
        if field:
            details["field"] = field
            
        super().__init__(
            message=message,
            error_code="PAYMENT_VALIDATION_ERROR",
            details=details
        )

class PaymentProcessingError(PaymentSystemError):
    """支付处理错误"""
    
    def __init__(self, message, transaction_id=None, reason=None):
        details = {}
        if transaction_id:
            details["transaction_id"] = transaction_id
        if reason:
            details["reason"] = reason
            
        super().__init__(
            message=message,
            error_code="PAYMENT_PROCESSING_ERROR",
            details=details
        )

class InsufficientFundsError(PaymentProcessingError):
    """余额不足错误"""
    
    def __init__(self, amount, current_balance, transaction_id=None):
        message = f"余额不足: 需要{amount}，当前余额{current_balance}"
        details = {
            "required_amount": amount,
            "current_balance": current_balance,
            "shortfall": amount - current_balance
        }
        
        super().__init__(
            message=message,
            transaction_id=transaction_id,
            reason="insufficient_funds"
        )

class CardDeclinedError(PaymentProcessingError):
    """卡片被拒绝错误"""
    
    def __init__(self, reason_code, transaction_id=None):
        reason_messages = {
            "expired": "卡片已过期",
            "stolen": "卡片挂失",
            "limit_exceeded": "超过信用额度"
        }
        
        message = f"卡片被拒绝: {reason_messages.get(reason_code, '未知原因')}"
        
        super().__init__(
            message=message,
            transaction_id=transaction_id,
            reason=reason_code
        )

# 使用层次化异常
def process_payment(amount, card_info, user_balance):
    """处理支付"""
    try:
        # 验证支付信息
        if not card_info.get("card_number"):
            raise PaymentValidationError(
                "卡号不能为空",
                field="card_number",
                payment_method="credit_card"
            )
        
        # 检查余额
        if user_balance < amount:
            raise InsufficientFundsError(
                amount=amount,
                current_balance=user_balance,
                transaction_id="TXN_12345"
            )
        
        # 模拟卡片验证
        if card_info.get("status") == "expired":
            raise CardDeclinedError(
                reason_code="expired",
                transaction_id="TXN_12345"
            )
        
        print("支付处理成功")
        return {"status": "success", "transaction_id": "TXN_12345"}
        
    except PaymentValidationError as e:
        print(f"支付验证失败: {e}")
        return e.to_dict()
    except InsufficientFundsError as e:
        print(f"余额不足: {e}")
        return e.to_dict()
    except CardDeclinedError as e:
        print(f"卡片被拒绝: {e}")
        return e.to_dict()
    except PaymentSystemError as e:
        print(f"支付系统错误: {e}")
        return e.to_dict()

# 测试支付处理
payment_cases = [
    {"amount": 100, "card_info": {"card_number": "1234"}, "balance": 200},
    {"amount": 100, "card_info": {"card_number": ""}, "balance": 200},
    {"amount": 300, "card_info": {"card_number": "1234"}, "balance": 200},
    {"amount": 100, "card_info": {"card_number": "1234", "status": "expired"}, "balance": 200},
]

for case in payment_cases:
    print(f"\n处理支付: 金额{case['amount']}, 余额{case['balance']}")
    result = process_payment(
        case["amount"], 
        case["card_info"], 
        case["balance"]
    )
    print(f"结果: {result}")

三、高级异常处理技巧

1. 异常链与上下文保留

def process_data_file(filename):
    """处理数据文件 - 演示异常链"""
    try:
        with open(filename, 'r') as file:
            data = file.read()
            numbers = [int(line.strip()) for line in data.split('\n') if line.strip()]
            return sum(numbers) / len(numbers)
    
    except FileNotFoundError as e:
        # 保留原始异常信息，添加上下文
        raise AppError(
            f"无法处理文件 {filename}",
            error_code="FILE_PROCESSING_ERROR",
            details={"filename": filename}
        ) from e
    
    except (ValueError, ZeroDivisionError) as e:
        # 转换异常类型，保留原始信息
        raise AppError(
            "文件内容格式错误",
            error_code="DATA_FORMAT_ERROR",
            details={"filename": filename, "original_error": str(e)}
        ) from e

def analyze_dataset(filenames):
    """分析数据集 - 多层异常处理"""
    results = []
    
    for filename in filenames:
        try:
            avg = process_data_file(filename)
            results.append({"filename": filename, "average": avg})
            
        except AppError as e:
            print(f"处理文件 {filename} 时发生错误: {e}")
            
            # 查看异常链
            if e.__cause__:
                print(f"根本原因: {e.__cause__}")
            
            # 记录错误但继续处理其他文件
            results.append({"filename": filename, "error": e.message})
            continue
    
    return results

# 测试异常链
test_files = ["data1.txt", "nonexistent.txt", "data2.txt", "invalid_data.txt"]

# 先创建测试文件
with open("data1.txt", "w") as f:
    f.write("10\n20\n30\n")

with open("data2.txt", "w") as f:
    f.write("5\n15\n25\n")

with open("invalid_data.txt", "w") as f:
    f.write("abc\ndef\n")  # 无效数字

print("分析数据集:")
results = analyze_dataset(test_files)

print("\n分析结果:")
for result in results:
    if "average" in result:
        print(f"{result['filename']}: 平均值 = {result['average']}")
    else:
        print(f"{result['filename']}: 错误 = {result['error']}")

2. 上下文管理器与异常处理

class DatabaseTransaction:
    """数据库事务上下文管理器"""
    
    def __init__(self, db_connection):
        self.db_connection = db_connection
        self.transaction_active = False
    
    def __enter__(self):
        self.transaction_active = True
        print("开始数据库事务")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.transaction_active = False
        
        if exc_type is None:
            # 没有异常，提交事务
            print("提交事务")
            return True
        else:
            # 有异常，回滚事务
            print(f"回滚事务，原因: {exc_val}")
            
            # 可以根据异常类型决定是否重新抛出
            if isinstance(exc_val, (ValueError, TypeError)):
                print("业务逻辑错误，已处理")
                return True  # 抑制异常
            else:
                print("系统错误，需要重新抛出")
                return False  # 重新抛出异常
    
    def execute_query(self, query):
        """执行查询"""
        if not self.transaction_active:
            raise RuntimeError("不在事务中")
        
        print(f"执行查询: {query}")
        return f"结果: {query}"

# 使用事务上下文管理器
def process_business_operation():
    """处理业务操作"""
    db_connection = "模拟数据库连接"
    
    try:
        with DatabaseTransaction(db_connection) as transaction:
            transaction.execute_query("INSERT INTO users VALUES ('Alice')")
            transaction.execute_query("UPDATE accounts SET balance = 100")
            # 模拟业务错误
            raise ValueError("业务规则验证失败")
            transaction.execute_query("COMMIT")
    
    except Exception as e:
        print(f"业务操作失败: {e}")

def process_critical_operation():
    """处理关键操作"""
    db_connection = "模拟数据库连接"
    
    try:
        with DatabaseTransaction(db_connection) as transaction:
            transaction.execute_query("INSERT INTO logs VALUES ('start')")
            # 模拟系统错误
            raise MemoryError("内存不足")
            transaction.execute_query("INSERT INTO logs VALUES ('end')")
    
    except Exception as e:
        print(f"关键操作失败: {e}")

# 测试不同的异常处理策略
print("=== 测试业务操作 ===")
process_business_operation()

print("\n=== 测试关键操作 ===")
process_critical_operation()

3. 重试机制与断路器模式

import time
import random
from functools import wraps

class RetryableError(AppError):
    """可重试的错误"""
    pass

class CircuitBreaker:
    """断路器模式"""
    
    def __init__(self, failure_threshold=3, recovery_timeout=10):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def can_execute(self):
        """检查是否允许执行"""
        if self.state == "OPEN":
            # 检查是否超过恢复超时
            if (time.time() - self.last_failure_time) > self.recovery_timeout:
                self.state = "HALF_OPEN"
                return True
            return False
        return True
    
    def record_success(self):
        """记录成功"""
        self.failures = 0
        self.state = "CLOSED"
    
    def record_failure(self):
        """记录失败"""
        self.failures += 1
        self.last_failure_time = time.time()
        
        if self.failures >= self.failure_threshold:
            self.state = "OPEN"

def retry_on_failure(max_retries=3, delay=1, backoff=2, exceptions=(Exception,)):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            current_delay = delay
            
            while retries <= max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    retries += 1
                    if retries > max_retries:
                        print(f"达到最大重试次数 {max_retries}，最终失败: {e}")
                        raise
                    
                    print(f"第 {retries} 次尝试失败: {e}，{current_delay}秒后重试")
                    time.sleep(current_delay)
                    current_delay *= backoff  # 指数退避
            
            return None
        return wrapper
    return decorator

def circuit_breaker_protected(circuit_breaker):
    """断路器保护装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if not circuit_breaker.can_execute():
                raise RetryableError("断路器已打开，拒绝执行")
            
            try:
                result = func(*args, **kwargs)
                circuit_breaker.record_success()
                return result
            except Exception as e:
                circuit_breaker.record_failure()
                if isinstance(e, RetryableError):
                    raise RetryableError(f"受保护操作失败: {e}") from e
                raise
        return wrapper
    return decorator

# 使用重试和断路器
@retry_on_failure(max_retries=3, delay=1, exceptions=(RetryableError,))
@circuit_breaker_protected(CircuitBreaker(failure_threshold=2, recovery_timeout=5))
def unreliable_remote_call():
    """模拟不可靠的远程调用"""
    if random.random() < 0.7:  # 70%概率失败
        raise RetryableError("远程服务暂时不可用")
    return "调用成功"

# 测试重试和断路器
print("测试重试机制和断路器:")
for i in range(10):
    try:
        result = unreliable_remote_call()
        print(f"尝试 {i+1}: {result}")
        time.sleep(0.5)  # 间隔一下
    except RetryableError as e:
        print(f"尝试 {i+1}: 最终失败 - {e}")
        time.sleep(2)  # 给断路器时间恢复

四、调试与日志记录

1. 结构化异常日志记录

import logging
import json
from datetime import datetime

class StructuredErrorLogger:
    """结构化错误日志记录器"""
    
    def __init__(self, log_file="errors.json"):
        self.logger = logging.getLogger("StructuredErrorLogger")
        self.logger.setLevel(logging.ERROR)
        
        # 创建JSON格式的处理器
        handler = logging.FileHandler(log_file)
        
        # 自定义JSON格式
        class JSONFormatter(logging.Formatter):
            def format(self, record):
                log_entry = {
                    "timestamp": datetime.now().isoformat(),
                    "level": record.levelname,
                    "message": record.getMessage(),
                    "exception": getattr(record, 'exception_info', None),
                    "context": getattr(record, 'context', {}),
                    "module": record.module,
                    "function": record.funcName,
                    "line": record.lineno
                }
                return json.dumps(log_entry, ensure_ascii=False)
        
        handler.setFormatter(JSONFormatter())
        self.logger.addHandler(handler)
    
    def log_error(self, error, context=None, exc_info=True):
        """记录错误"""
        extra = {
            'context': context or {},
            'exception_info': str(error) if exc_info else None
        }
        
        self.logger.error(
            f"应用程序错误: {error}",
            extra=extra,
            exc_info=exc_info
        )

def error_handler(func):
    """错误处理装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger = StructuredErrorLogger()
        
        try:
            return func(*args, **kwargs)
        except AppError as e:
            # 业务异常，记录并返回友好错误
            context = {
                "function": func.__name__,
                "args": str(args),
                "error_code": e.error_code,
                "details": e.details
            }
            logger.log_error(e, context=context, exc_info=False)
            
            # 返回结构化错误响应
            return e.to_dict()
        except Exception as e:
            # 系统异常，详细记录
            context = {
                "function": func.__name__,
                "args": str(args),
                "timestamp": datetime.now().isoformat()
            }
            logger.log_error(e, context=context, exc_info=True)
            
            # 返回通用错误响应
            return {
                "error": "系统内部错误",
                "code": "INTERNAL_ERROR",
                "timestamp": context["timestamp"]
            }
    return wrapper

# 使用错误处理装饰器
@error_handler
def process_user_request(user_data):
    """处理用户请求"""
    # 模拟业务逻辑
    if not user_data.get("user_id"):
        raise ValidationError("用户ID不能为空", field="user_id")
    
    if user_data.get("action") == "dangerous":
        raise Exception("危险操作被阻止")
    
    return {"status": "success", "data": "处理完成"}

# 测试错误处理
test_requests = [
    {"user_id": 123, "action": "safe"},
    {"user_id": None, "action": "safe"},
    {"user_id": 123, "action": "dangerous"}
]

print("处理用户请求:")
for request in test_requests:
    result = process_user_request(request)
    print(f"请求: {request} -> 结果: {result}")

2. 调试友好的异常信息

def create_debuggable_exception(message, debug_info=None):
    """创建包含调试信息的异常"""
    class DebuggableError(Exception):
        def __init__(self, message, debug_info=None):
            self.message = message
            self.debug_info = debug_info or {}
            self.timestamp = datetime.now()
            super().__init__(self.message)
        
        def __str__(self):
            base_message = super().__str__()
            debug_str = json.dumps(self.debug_info, indent=2, default=str)
            return f"{base_message}\n调试信息:\n{debug_str}"
        
        def to_debug_dict(self):
            """转换为调试字典"""
            return {
                "message": self.message,
                "debug_info": self.debug_info,
                "timestamp": self.timestamp.isoformat(),
                "type": self.__class__.__name__
            }
    
    return DebuggableError(message, debug_info)

def debug_wrapper(func):
    """调试包装器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 增强异常信息
            debug_info = {
                "function": func.__name__,
                "args": str(args)[:100],  # 限制长度
                "kwargs": str(kwargs)[:100],
                "local_vars": {k: str(v)[:50] for k, v in locals().items() 
                              if not k.startswith('_') and k != 'e'},
                "call_stack": [f"{f.filename}:{f.lineno} in {f.function}" 
                              for f in inspect.stack()[1:6]]  # 最近5个调用
            }
            
            raise create_debuggable_exception(str(e), debug_info) from e
    return wrapper

# 使用调试包装器
@debug_wrapper
def complex_calculation(data):
    """复杂计算函数"""
    result = data["a"] / data["b"]  # 可能除零
    processed = result + data["c"]["d"]  # 可能键错误
    return processed * data.get("multiplier", 1)

# 测试调试功能
test_data = [
    {"a": 10, "b": 2, "c": {"d": 5}},  # 正常
    {"a": 10, "b": 0, "c": {"d": 5}},  # 除零错误
    {"a": 10, "b": 2, "c": {}},        # 键错误
]

for data in test_data:
    try:
        result = complex_calculation(data)
        print(f"数据: {data} -> 结果: {result}")
    except Exception as e:
        print(f"数据: {data} -> 错误: {e}")
        print("-" * 50)

五、最佳实践总结

1. 异常处理原则检查清单

class ExceptionHandlingGuidelines:
    """异常处理指导原则"""
    
    @staticmethod
    def validate_exception_handling(code_snippet):
        """验证异常处理代码"""
        warnings = []
        
        # 检查1: 避免裸露的except
        if "except:" in code_snippet and "except Exception:" not in code_snippet:
            warnings.append("避免使用裸露的except语句，应该明确指定异常类型")
        
        # 检查2: 不要静默忽略异常
        if "except" in code_snippet and "pass" in code_snippet:
            warnings.append("不要静默忽略异常，至少应该记录日志")
        
        # 检查3: 检查资源清理
        if "open(" in code_snippet and "with " not in code_snippet:
            warnings.append("文件操作应该使用with语句确保正确关闭")
        
        # 检查4: 避免过度捕获
        if code_snippet.count("except") > 3:
            warnings.append("异常捕获可能过于复杂，考虑重构")
        
        return warnings
    
    @staticmethod
    def get_best_practices():
        """返回异常处理最佳实践"""
        return {
            "specificity": "捕获具体的异常类型，而不是通用的Exception",
            "context": "在异常消息中包含足够的上下文信息",
            "recovery": "提供有意义的错误恢复策略",
            "logging": "记录异常信息用于调试和监控",
            "propagation": "在适当的层级处理异常，不要过早捕获",
            "custom_exceptions": "为业务逻辑定义自定义异常类型",
            "resource_management": "使用上下文管理器管理资源",
            "user_experience": "为用户提供友好的错误消息"
        }

# 使用指导原则
code_example = """
try:
    file = open("data.txt")
    data = file.read()
    result = int(data)
except:
    pass
"""

warnings = ExceptionHandlingGuidelines.validate_exception_handling(code_example)
print("代码检查警告:")
for warning in warnings:
    print(f"- {warning}")

print("\n异常处理最佳实践:")
practices = ExceptionHandlingGuidelines.get_best_practices()
for key, practice in practices.items():
    print(f"- {key}: {practice}")

2. 异常处理策略模板

def create_robust_function(func_name, operation_description):
    """创建健壮函数的模板"""
    template = f'''
def {func_name}(*args, **kwargs):
    """
    {operation_description}
    
    异常处理策略:
    - ValidationError: 输入数据验证失败
    - DatabaseError: 数据库操作失败  
    - NetworkError: 网络通信失败
    - 其他异常: 系统内部错误
    """
    logger = StructuredErrorLogger()
    
    try:
        # 主要业务逻辑
        result = _execute_{func_name}_core(*args, **kwargs)
        return {{"status": "success", "data": result}}
        
    except ValidationError as e:
        logger.log_error(e, context={{"function": "{func_name}"}}, exc_info=False)
        return e.to_dict()
        
    except (DatabaseError, NetworkError) as e:
        logger.log_error(e, context={{"function": "{func_name}"}}, exc_info=True)
        return e.to_dict()
        
    except Exception as e:
        logger.log_error(e, context={{"function": "{func_name}"}}, exc_info=True)
        return {{
            "error": "系统内部错误",
            "code": "INTERNAL_ERROR",
            "request_id": "REQ_12345"  # 实际中应该生成唯一ID
        }}
    
def _execute_{func_name}_core(*args, **kwargs):
    """核心业务逻辑实现"""
    # 在这里实现具体的业务逻辑
    raise NotImplementedError("需要实现核心逻辑")
'''

    return template

# 生成函数模板
user_creation_template = create_robust_function(
    "create_user", 
    "创建新用户账户"
)

print("生成的函数模板:")
print(user_creation_template)

总结

Python异常处理是一个多层次、系统化的工程，关键要点包括：

1. 基础语法掌握：熟练使用try-except-else-finally，理解不同代码块的执行顺序
2. 异常类型识别：区分业务异常和系统异常，采用不同的处理策略
3. 自定义异常体系：建立层次化的异常类，提供丰富的错误上下文
4. 资源管理：使用上下文管理器确保资源正确释放
5. 重试与容错：实现重试机制和断路器模式提高系统韧性
6. 日志与监控：结构化记录异常信息，便于调试和问题追踪
7. 用户体验：为用户提供清晰友好的错误信息

通过系统化的异常处理策略，可以构建出既健壮又易于维护的Python应用程序。记住，好的异常处理不是阻止所有错误，而是优雅地处理不可避免的错误。