Day 31 - raise 语句与自定义异常

raise 语句基础

raise 语句用于显式地抛出异常。在 Python 中，你可以抛出任何继承自 BaseException 的对象，但通常我们应该抛出 Exception 的子类。

# 基本语法
raise Exception("错误信息")

# 重新抛出当前捕获的异常
try:
    raise ValueError("原始错误")
except ValueError:
    print("处理异常")
    raise  # 重新抛出

# 使用 raise from 链接异常
try:
    raise ValueError("原始错误")
except ValueError as e:
    raise RuntimeError("新错误") from e

常见的内置异常

Python 提供了一系列内置异常，按层次结构组织：

BaseException ├── SystemExit ├── KeyboardInterrupt ├── GeneratorExit └── Exception ├── StopIteration ├── ArithmeticError │ ├── FloatingPointError │ ├── OverflowError │ └── ZeroDivisionError ├── LookupError │ ├── IndexError │ └── KeyError ├── OSError (IOError) │ ├── FileNotFoundError │ ├── PermissionError │ └── TimeoutError ├── TypeError ├── ValueError ├── RuntimeError └── ...
# 触发各种内置异常
print(10 / 0)  # ZeroDivisionError
print(int("abc"))  # ValueError
print([1,2][10])  # IndexError
print({}["key"])  # KeyError
open("nonexistent.txt")  # FileNotFoundError

自定义异常类

自定义异常应该继承自 Exception 或其子类。通常，自定义异常类比较简单，只需要定义构造函数和异常消息。

class MyError(Exception):
    """自定义错误"""
    pass

raise MyError("这是一个自定义错误")

带额外信息的异常

class ValidationError(Exception):
    """验证错误：包含字段名和错误信息"""
    
    def __init__(self, field, message):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")

raise ValidationError("email", "邮箱格式不正确")
# ValidationError: email: 邮箱格式不正确

完整的企业级异常设计

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):
        parts = [self.message]
        if self.error_code:
            parts.append(f"[错误码: {self.error_code}]")
        if self.details:
            parts.append(f"详情: {self.details}")
        return " ".join(parts)

class ValidationError(AppError):
    """验证错误"""
    def __init__(self, field, message, value=None):
        details = {"field": field, "value": value}
        super().__init__(f"验证失败: {message}", "VALIDATION_ERROR", details)

class DatabaseError(AppError):
    """数据库错误"""
    pass

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

# 使用
try:
    raise ValidationError("username", "用户名不能为空", value="")
except AppError as e:
    print(f"错误类型: {type(e).__name__}")
    print(f"错误码: {e.error_code}")
    print(f"详情: {e.details}")

异常类的设计原则

  1. 继承层次合理：根据异常的语义和用途选择合适的基类
  2. 提供有用的信息：异常消息应该清楚描述问题
  3. 包含上下文信息：使用属性存储相关上下文
  4. 遵循命名约定：异常类名以 Error 结尾
  5. 不过度具体化：避免创建过多细粒度的异常类
# 不好的设计：过于具体
class UsernameEmptyError(Exception): pass
class UsernameTooShortError(Exception): pass
class UsernameTooLongError(Exception): pass

# 好的设计：统一的验证错误
class ValidationError(Exception):
    def __init__(self, field, rule, value=None):
        self.field = field
        self.rule = rule
        self.value = value
        super().__init__(f"{field} 违反规则 {rule}，值: {value}")

raise ValidationError("username", "not_empty", value=None)
raise ValidationError("username", "min_length:3", value="ab")
raise ValidationError("username", "max_length:20", value="a" * 25)

raise 语句的多种形式

1. raise 单独使用

try:
    raise ValueError("错误信息")
except ValueError:
    print("捕获了 ValueError")
    raise  # 重新抛出，不带参数

2. raise Exception

raise ValueError("错误信息")

3. raise Exception from original

try:
    raise TypeError("原始错误")
except TypeError as e:
    raise RuntimeError("新错误") from e

4. raise Exception from None

raise RuntimeError("新错误") from None

异常链详解

异常链允许在一个异常中保留另一个异常的上下文。

# 显式异常链（from）
try:
    raise ValueError("原始错误")
except ValueError as e:
    raise RuntimeError("中间层错误") from e

# 输出：
# Traceback (most recent call last):
#   File "...", line 2, in <module>
#     raise ValueError("原始错误")
# ValueError: 原始错误
# 
# The above exception was the direct cause of the following exception:
# Traceback (most recent call last):
#   File "...", line 4, in <module>
#     raise RuntimeError("中间层错误") from e
# RuntimeError: 中间层错误

# 隐式异常链
try:
    raise ValueError("原始错误")
except ValueError:
    raise RuntimeError("新错误")  # 没有 from，隐式链接

# 输出会显示：
# During handling of the above exception, another exception occurred:

断言与异常

断言（assert）用于开发过程中检查不可能发生的情况，断言失败会抛出 AssertionError。

# 断言语法
assert condition, "错误信息"

# 示例
def divide(a, b):
    assert b != 0, "除数不能为零"
    return a / b

divide(10, 2)  # 正常
divide(10, 0)  # AssertionError: 除数不能为零

断言 vs 异常：

  • 断言用于检测程序错误（不应该发生的情况）
  • 异常用于处理运行时可能发生的错误情况
  • 断言可以在 Python 启动时使用 -O 参数禁用
# 不适合用断言的情况
def get_user(user_id):
    user = db.get(user_id)  # 数据库可能不存在
    assert user is not None  # 不好：应该用异常
    return user

# 适合用断言的情况
def calculate_triangle_area(base, height):
    assert base > 0 and height > 0  # 不可能为负数
    return base * height / 2

异常处理的反模式

反模式1：空的 except 块

# 不好：隐藏了所有错误
try:
    risky_operation()
except:
    pass

# 好：捕获特定异常
try:
    risky_operation()
except SpecificError as e:
    logger.error(f"操作失败: {e}")
    raise

反模式2：捕获所有异常

# 不好：过于宽泛
try:
    operation()
except Exception:
    handle_error()

# 好：分层处理
try:
    operation()
except ValidationError as e:
    return bad_request(str(e))
except DatabaseError as e:
    return server_error("数据库错误")
except Exception as e:
    logger.critical(f"未知错误: {e}")
    return server_error("系统错误")

反模式3：异常用于控制流

# 不好：用异常控制流程
try:
    result = items[current_index]
    current_index += 1
except IndexError:
    current_index = 0
    result = items[0]

# 好：使用正常的控制流
if current_index < len(items):
    result = items[current_index]
    current_index += 1
else:
    current_index = 0
    result = items[0]

异常与调试

使用 traceback 模块

import traceback

try:
    raise ValueError("测试错误")
except ValueError:
    # 打印完整的 traceback
    traceback.print_exc()
    
    # 获取 traceback 字符串
    tb_str = traceback.format_exc()
    print(f"捕获的异常: {tb_str}")

使用 logging 记录异常

import logging

logger = logging.getLogger(__name__)

try:
    operation()
except Exception as e:
    logger.exception("操作失败")  # 自动记录完整 traceback
    # 或
    logger.error("操作失败: %s", str(e), exc_info=True)

练习题

练习 1：实现 API 异常

class APIError(Exception):
    """API 错误基类"""
    
    def __init__(self, message, status_code=500, response=None):
        self.message = message
        self.status_code = status_code
        self.response = response
        super().__init__(self.message)

class AuthenticationError(APIError):
    """认证错误"""
    def __init__(self, message="认证失败"):
        super().__init__(message, status_code=401)

class AuthorizationError(APIError):
    """授权错误"""
    def __init__(self, message="权限不足"):
        super().__init__(message, status_code=403)

class NotFoundError(APIError):
    """资源不存在错误"""
    def __init__(self, resource, resource_id):
        message = f"{resource} 不存在: {resource_id}"
        super().__init__(message, status_code=404)
        self.resource = resource
        self.resource_id = resource_id

class RateLimitError(APIError):
    """请求频率限制错误"""
    def __init__(self, retry_after=60):
        message = f"请求过于频繁，请在 {retry_after} 秒后重试"
        super().__init__(message, status_code=429)
        self.retry_after = retry_after

def api_call(endpoint):
    """模拟 API 调用"""
    if endpoint == "/auth/login":
        raise AuthenticationError("Token 已过期")
    elif endpoint == "/admin/users":
        raise AuthorizationError("需要管理员权限")
    elif endpoint == "/users/999":
        raise NotFoundError("用户", 999)
    elif endpoint == "/api/data":
        raise RateLimitError(30)
    return {"data": "success"}

# 测试
endpoints = ["/auth/login", "/admin/users", "/users/999", "/api/data", "/api/valid"]

for endpoint in endpoints:
    try:
        result = api_call(endpoint)
        print(f"{endpoint}: {result}")
    except APIError as e:
        print(f"{endpoint}:")
        print(f"  状态码: {e.status_code}")
        print(f"  消息: {e.message}")
        if hasattr(e, 'retry_after'):
            print(f"  重试时间: {e.retry_after}秒")
        print()

练习 2：验证器

class ValidationError(Exception):
    """验证错误"""
    
    def __init__(self, field, rule, value=None):
        self.field = field
        self.rule = rule
        self.value = value
        super().__init__(f"{field}: {rule} (当前值: {value})")

def validate_required(value, field_name):
    if value is None or (isinstance(value, str) and not value.strip()):
        raise ValidationError(field_name, "必填字段")

def validate_length(value, field_name, min_len=None, max_len=None):
    if value is None:
        return
    length = len(value)
    if min_len is not None and length < min_len:
        raise ValidationError(field_name, f"最小长度: {min_len}", value)
    if max_len is not None and length > max_len:
        raise ValidationError(field_name, f"最大长度: {max_len}", value)

def validate_range(value, field_name, min_val=None, max_val=None):
    if value is None:
        return
    if min_val is not None and value < min_val:
        raise ValidationError(field_name, f"最小值: {min_val}", value)
    if max_val is not None and value > max_val:
        raise ValidationError(field_name, f"最大值: {max_val}", value)

def validate_email(email, field_name):
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    if not re.match(pattern, email):
        raise ValidationError(field_name, "无效的邮箱格式", email)

def validate_user_data(data):
    """验证用户数据"""
    errors = []
    
    for field, rules in data.items():
        value = rules.get('value')
        
        # 必填检查
        if rules.get('required', False):
            try:
                validate_required(value, field)
            except ValidationError as e:
                errors.append(e)
        
        # 长度检查
        if 'min_length' in rules or 'max_length' in rules:
            try:
                validate_length(value, field, 
                              rules.get('min_length'), 
                              rules.get('max_length'))
            except ValidationError as e:
                errors.append(e)
        
        # 范围检查
        if 'min' in rules or 'max' in rules:
            try:
                validate_range(value, field, rules.get('min'), rules.get('max'))
            except ValidationError as e:
                errors.append(e)
        
        # 邮箱检查
        if rules.get('type') == 'email':
            try:
                validate_email(value, field)
            except ValidationError as e:
                errors.append(e)
    
    if errors:
        raise ValidationError("user_data", "验证失败")
    
    return True

# 测试
test_data = {
    "username": {"value": "john", "required": True, "min_length": 3, "max_length": 20},
    "email": {"value": "john@example.com", "required": True, "type": "email"},
    "age": {"value": 25, "min": 0, "max": 150},
    "bio": {"value": "", "max_length": 500}
}

try:
    validate_user_data(test_data)
    print("验证通过！")
except ValidationError as e:
    print(f"验证失败: {e}")

# 测试无效数据
invalid_data = {
    "username": {"value": "jo", "required": True, "min_length": 3},
    "email": {"value": "invalid-email", "type": "email"},
    "age": {"value": -5, "min": 0}
}

try:
    validate_user_data(invalid_data)
except ValidationError as e:
    print(f"验证失败: {e}")
    if hasattr(e, 'field'):
        print(f"字段: {e.field}, 规则: {e.rule}, 值: {e.value}")

练习 3：上下文管理器与异常

class Transaction:
    """事务上下文管理器"""
    
    def __init__(self, connection):
        self.conn = connection
        self.committed = False
    
    def __enter__(self):
        print("开始事务")
        self.conn.begin()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            print(f"事务失败: {exc_val}")
            print("执行回滚")
            self.conn.rollback()
            return False  # 不吞没异常
        else:
            print("提交事务")
            self.conn.commit()
            self.committed = True
            return True  # 吞没异常（如果有）
    
    def execute(self, sql):
        print(f"执行 SQL: {sql}")
        self.conn.execute(sql)

class MockConnection:
    def begin(self):
        print("  [连接] 开始事务")
    
    def commit(self):
        print("  [连接] 提交事务")
    
    def rollback(self):
        print("  [连接] 回滚事务")
    
    def execute(self, sql):
        print(f"  [连接] 执行: {sql}")

# 测试成功场景
print("=== 测试成功场景 ===")
conn = MockConnection()
try:
    with Transaction(conn) as tx:
        tx.execute("INSERT INTO users VALUES (1, '张三')")
        tx.execute("UPDATE accounts SET balance = balance - 100")
        print(f"事务提交状态: {tx.committed}")
except Exception as e:
    print(f"外部捕获异常: {e}")

print("\n=== 测试失败场景 ===")
conn2 = MockConnection()
try:
    with Transaction(conn2) as tx:
        tx.execute("INSERT INTO orders VALUES (1, '商品')")
        tx.execute("DELETE FROM inventory WHERE quantity < 0")  # 这会失败
except RuntimeError as e:
    print(f"外部捕获异常: {e}")

总结

raise 语句和自定义异常是 Python 异常处理的重要组成部分：

  1. raise 语句：显式抛出异常，支持异常链
  2. 自定义异常：继承 Exception 或其子类
  3. 异常链：raise ... from e 和 raise ... from None
  4. 断言：用于开发时检查不可能的情况
  5. 异常设计原则：提供有用的信息，包含上下文
  6. 反模式：避免空 except、过度宽泛的捕获等

在下一节中，我们将学习 Python 的模块系统：import、from、as。