Python异常链详解 - raise from与异常上下文追踪

def load_config(filepath):
    try:
        with open(filepath) as f:
            import json
            return json.load(f)
    except FileNotFoundError as e:
        raise ValueError(f"配置文件不存在：{filepath}") from e

try:
    load_config("missing.json")
except ValueError as e:
    print(f"捕获的异常：{e}")
    print(f"__cause__：{e.__cause__}")
    print(f"__cause__类型：{type(e.__cause__).__name__}")

# 输出示例：
# Traceback (most recent call last):
#   File "config.py", line 4, in load_config
#     with open(filepath) as f:
# FileNotFoundError: [Errno 2] No such file or directory: 'missing.json'
#
# The above exception was the direct cause of the following exception:
#
# Traceback (most recent call last):
#   File "config.py", line 12, in <module>
#     load_config("missing.json")
# ValueError: 配置文件不存在：missing.json

def divide(a, b):
    try:
        result = a / b
    except ZeroDivisionError:
        # 隐式异常链：没有使用from
        raise ValueError("除法运算失败")

try:
    divide(10, 0)
except ValueError as e:
    print(f"异常：{e}")
    print(f"__context__：{e.__context__}")
    print(f"__cause__：{e.__cause__}")  # None，因为没有使用from

def print_exception_chain(exc, indent=0):
    """递归打印异常链"""
    prefix = "  " * indent
    print(f"{prefix}[{type(exc).__name__}] {exc}")
    
    if exc.__cause__:
        print(f"{prefix}  └─ __cause__:")
        print_exception_chain(exc.__cause__, indent + 2)
    elif exc.__context__:
        print(f"{prefix}  └─ __context__:")
        print_exception_chain(exc.__context__, indent + 2)

# 构建多层异常链
def deep_operation():
    try:
        try:
            int("not_a_number")  # 抛出 ValueError
        except ValueError as e1:
            raise RuntimeError("解析失败") from e1
    except RuntimeError as e2:
        raise SystemError("系统异常") from e2

try:
    deep_operation()
except SystemError as e:
    print_exception_chain(e)

class DatabaseConnectionError(Exception):
    pass

class QueryError(Exception):
    pass

def connect_to_db(host, port):
    """模拟数据库连接"""
    import socket
    try:
        sock = socket.socket()
        sock.settimeout(1)
        sock.connect((host, port))
        return sock
    except socket.timeout as e:
        raise DatabaseConnectionError(f"连接超时：{host}:{port}") from e
    except ConnectionRefusedError as e:
        raise DatabaseConnectionError(f"连接被拒绝：{host}:{port}") from e

def execute_query(connection, sql):
    """模拟查询执行"""
    try:
        # 模拟查询失败
        raise ValueError(f"SQL语法错误：{sql}")
    except ValueError as e:
        raise QueryError(f"查询执行失败") from e

def get_user(user_id):
    """获取用户信息（完整流程）"""
    conn = connect_to_db("localhost", 5432)
    try:
        return execute_query(conn, f"SELECT * FROM users WHERE id={user_id}")
    finally:
        conn.close()

try:
    get_user(1)
except (DatabaseConnectionError, QueryError) as e:
    print(f"操作失败：{e}")
    if e.__cause__:
        print(f"底层原因：{type(e.__cause__).__name__}: {e.__cause__}")

class UserService:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def find_by_email(self, email):
        try:
            # 模拟数据库错误
            raise Exception("数据库内部错误")
        except Exception:
            # 使用 from None 隐藏技术细节
            # 用户不需要知道底层是数据库错误
            raise LookupError(f"未找到邮箱为 {email} 的用户") from None

service = UserService(None)
try:
    service.find_by_email("test@example.com")
except LookupError as e:
    print(f"用户友好的错误信息：{e}")
    print(f"__cause__: {e.__cause__}")  # None - 技术细节已隐藏

def process_data(raw_data):
    """处理原始数据"""
    try:
        import json
        parsed = json.loads(raw_data)
    except json.JSONDecodeError as e:
        parse_error = e
    
    # 在except块外部使用raise from
    if 'parse_error' in locals():
        raise ValueError("数据格式无效") from parse_error
    
    return parsed

try:
    process_data("{invalid}")
except ValueError as e:
    print(f"异常：{e}")
    print(f"原因：{e.__cause__}")