Python异常处理

学习目标

通过本集的学习，你将能够：

• 理解什么是异常
• 使用 try/except 捕获和处理异常
• 使用 finally 执行清理代码
• 抛出异常
• 创建和使用自定义异常

1. 什么是异常？

异常是程序运行时发生的错误，会中断程序的正常执行流程。

1.1 常见的异常类型

# ZeroDivisionError: 除零错误
try:
    result = 10 / 0
except ZeroDivisionError:
    print("不能除以零")

# ValueError: 值错误
try:
    num = int("abc")
except ValueError:
    print("无法转换为整数")

# TypeError: 类型错误
try:
    result = "10" + 5
except TypeError:
    print("类型不匹配")

# IndexError: 索引错误
try:
    lst = [1, 2, 3]
    print(lst[10])
except IndexError:
    print("索引超出范围")

# KeyError: 键错误
try:
    d = {"a": 1}
    print(d["b"])
except KeyError:
    print("键不存在")

# FileNotFoundError: 文件不存在
try:
    with open("nonexistent.txt", "r") as f:
        content = f.read()
except FileNotFoundError:
    print("文件不存在")

异常的ASCII图：

程序正常执行
    │
    ▼
发生错误 → 抛出异常
    │
    ▼
查找异常处理器
    │
    ├─ 找到 → 处理异常 → 继续执行
    │
    └─ 未找到 → 程序崩溃

1.2 异常的层级结构

BaseException
 ├── SystemExit
 ├── KeyboardInterrupt
 ├── GeneratorExit
 └── Exception
      ├── StopIteration
      ├── ArithmeticError
      │    ├── FloatingPointError
      │    ├── OverflowError
      │    └── ZeroDivisionError
      ├── AssertionError
      ├── AttributeError
      ├── BufferError
      ├── EOFError
      ├── ImportError
      ├── LookupError
      │    ├── IndexError
      │    └── KeyError
      ├── MemoryError
      ├── NameError
      ├── OSError
      │    ├── FileNotFoundError
      │    └── ...
      ├── RuntimeError
      ├── SyntaxError
      ├── TypeError
      └── ValueError

2. try/except 语句

2.1 基本用法

try:
    # 可能出错的代码
    x = int(input("请输入一个数字: "))
    result = 10 / x
    print(f"结果: {result}")
except ValueError:
    print("输入的不是有效数字")
except ZeroDivisionError:
    print("不能除以零")

2.2 捕获多个异常

try:
    x = int(input("请输入一个数字: "))
    y = int(input("请输入另一个数字: "))
    result = x / y
    print(f"结果: {result}")
except (ValueError, ZeroDivisionError) as e:
    print(f"发生错误: {e}")

2.3 获取异常信息

try:
    10 / 0
except ZeroDivisionError as e:
    print(f"异常类型: {type(e)}")
    print(f"异常信息: {e}")
    print(f"异常参数: {e.args}")

2.4 捕获所有异常

try:
    # 一些代码
    10 / 0
except Exception as e:
    print(f"捕获到异常: {e}")

# 注意：不推荐使用裸 except，因为它会捕获包括 KeyboardInterrupt 在内的所有异常
try:
    10 / 0
except:
    print("发生了某个异常")

2.5 else 子句

try:
    x = int(input("请输入一个数字: "))
except ValueError:
    print("输入无效")
else:
    # 没有异常时执行
    print(f"你输入的是: {x}")
    print(f"平方是: {x * x}")

3. finally 子句

finally 子句中的代码无论是否发生异常都会执行。

3.1 基本用法

try:
    f = open("example.txt", "r")
    content = f.read()
    print(content)
except FileNotFoundError:
    print("文件不存在")
finally:
    # 无论是否出错都会执行
    print("清理工作...")
    # 如果文件打开了，关闭它
    if 'f' in locals() and not f.closed:
        f.close()

3.2 文件操作的完整示例

def read_file_safely(filename):
    f = None
    try:
        f = open(filename, "r", encoding="utf-8")
        return f.read()
    except FileNotFoundError:
        print(f"文件不存在: {filename}")
        return None
    except PermissionError:
        print(f"没有权限读取文件: {filename}")
        return None
    finally:
        if f is not None:
            f.close()
            print("文件已关闭")

# 使用 with 语句（更简洁）
def read_file_with_with(filename):
    try:
        with open(filename, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        print(f"文件不存在: {filename}")
        return None

try-except-else-finally 的ASCII图：

try:
    代码块
    │
    ├─ 正常 → else 代码块
    │           │
    │           └─→ finally 代码块
    │
    └─ 异常 → except 代码块
                │
                └─→ finally 代码块

4. 抛出异常

使用 raise 语句可以主动抛出异常。

4.1 抛出内置异常

def divide(a, b):
    if b == 0:
        raise ZeroDivisionError("除数不能为零")
    return a / b

try:
    result = divide(10, 0)
except ZeroDivisionError as e:
    print(f"错误: {e}")

# 验证年龄
def validate_age(age):
    if not isinstance(age, int):
        raise TypeError("年龄必须是整数")
    if age < 0:
        raise ValueError("年龄不能为负数")
    if age > 150:
        raise ValueError("年龄不合理")
    return True

try:
    validate_age(-5)
except (TypeError, ValueError) as e:
    print(f"验证失败: {e}")

4.2 重新抛出异常

def process_data(data):
    try:
        # 处理数据
        result = int(data)
    except ValueError as e:
        print("记录错误日志...")
        raise  # 重新抛出异常

try:
    process_data("abc")
except ValueError:
    print("处理失败")

5. 自定义异常

通过继承 Exception 类创建自定义异常。

5.1 基本自定义异常

class ValidationError(Exception):
    """验证错误"""
    pass

def validate_username(username):
    if len(username) < 3:
        raise ValidationError("用户名至少需要3个字符")
    if not username.isalnum():
        raise ValidationError("用户名只能包含字母和数字")
    return True

try:
    validate_username("ab")
except ValidationError as e:
    print(f"验证错误: {e}")

5.2 带属性的自定义异常

class BankAccountError(Exception):
    """银行账户错误基类"""
    def __init__(self, message, account_id=None):
        super().__init__(message)
        self.account_id = account_id
        self.message = message

class InsufficientFundsError(BankAccountError):
    """余额不足错误"""
    def __init__(self, message, account_id, balance, amount):
        super().__init__(message, account_id)
        self.balance = balance
        self.amount = amount

class NegativeAmountError(BankAccountError):
    """金额为负错误"""
    pass

class BankAccount:
    def __init__(self, account_id, balance=0):
        self.account_id = account_id
        self.balance = balance
    
    def withdraw(self, amount):
        if amount < 0:
            raise NegativeAmountError(
                "取款金额不能为负数",
                self.account_id
            )
        if amount > self.balance:
            raise InsufficientFundsError(
                "余额不足",
                self.account_id,
                self.balance,
                amount
            )
        self.balance -= amount
        return amount

# 使用
account = BankAccount("ACC001", 1000)

try:
    account.withdraw(1500)
except InsufficientFundsError as e:
    print(f"错误: {e.message}")
    print(f"账户: {e.account_id}")
    print(f"余额: {e.balance}")
    print(f"尝试取款: {e.amount}")

try:
    account.withdraw(-100)
except NegativeAmountError as e:
    print(f"错误: {e.message}")

5.3 异常层级

class AppError(Exception):
    """应用程序错误基类"""
    pass

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

class ConnectionError(DatabaseError):
    """连接错误"""
    pass

class QueryError(DatabaseError):
    """查询错误"""
    pass

# 使用
try:
    raise ConnectionError("无法连接数据库")
except DatabaseError as e:
    print(f"数据库错误: {e}")
except AppError as e:
    print(f"应用错误: {e}")

6. 异常处理最佳实践

6.1 具体异常 vs 宽泛异常

# 好：捕获具体异常
try:
    f = open("file.txt", "r")
    x = int(f.readline())
except FileNotFoundError:
    print("文件不存在")
except ValueError:
    print("无法转换为整数")
finally:
    f.close()

# 不好：捕获过于宽泛
try:
    f = open("file.txt", "r")
    x = int(f.readline())
except:
    print("发生了错误")
finally:
    f.close()

6.2 不要忽略异常

# 不好：忽略异常
try:
    result = 10 / 0
except:
    pass

# 好：至少记录日志
import logging
try:
    result = 10 / 0
except Exception as e:
    logging.error(f"发生错误: {e}")

# 更好：处理异常
try:
    result = 10 / 0
except ZeroDivisionError:
    result = float('inf')  # 或其他合理的默认值

7. 实用案例

7.1 案例1：用户输入验证器

# input_validator.py

class InputError(Exception):
    """输入错误基类"""
    pass

class EmptyInputError(InputError):
    """空输入错误"""
    pass

class InvalidRangeError(InputError):
    """范围错误"""
    def __init__(self, message, value, min_val, max_val):
        super().__init__(message)
        self.value = value
        self.min_val = min_val
        self.max_val = max_val

def get_input(prompt, validator=None):
    """获取并验证用户输入"""
    while True:
        try:
            user_input = input(prompt).strip()
            
            if not user_input:
                raise EmptyInputError("输入不能为空")
            
            if validator:
                user_input = validator(user_input)
            
            return user_input
        
        except InputError as e:
            print(f"错误: {e}")
            print("请重试")

def validate_integer(input_str):
    """验证整数"""
    try:
        return int(input_str)
    except ValueError:
        raise InputError("必须输入整数")

def validate_range(min_val, max_val):
    """创建范围验证器"""
    def validator(value):
        num = validate_integer(value)
        if not (min_val <= num <= max_val):
            raise InvalidRangeError(
                f"必须在 {min_val} 到 {max_val} 之间",
                num, min_val, max_val
            )
        return num
    return validator

# 使用
print("=== 用户输入验证器 ===")

try:
    name = get_input("请输入姓名: ")
    age = get_input("请输入年龄 (1-150): ", validate_range(1, 150))
    score = get_input("请输入分数 (0-100): ", validate_range(0, 100))
    
    print("\n输入信息:")
    print(f"姓名: {name}")
    print(f"年龄: {age}")
    print(f"分数: {score}")
    
except KeyboardInterrupt:
    print("\n\n程序被用户中断")

7.2 案例2：配置文件加载器

# config_loader.py

import json
from pathlib import Path

class ConfigError(Exception):
    """配置错误基类"""
    pass

class ConfigNotFoundError(ConfigError):
    """配置文件不存在"""
    pass

class ConfigParseError(ConfigError):
    """配置解析错误"""
    pass

class ConfigValidationError(ConfigError):
    """配置验证错误"""
    pass

class ConfigLoader:
    """配置文件加载器"""
    
    REQUIRED_FIELDS = ["database", "server", "log_level"]
    VALID_LOG_LEVELS = ["DEBUG", "INFO", "WARNING", "ERROR"]
    
    def __init__(self, config_path):
        self.config_path = Path(config_path)
        self.config = None
    
    def load(self):
        """加载配置"""
        self._check_file_exists()
        self._parse_config()
        self._validate_config()
        return self.config
    
    def _check_file_exists(self):
        """检查文件是否存在"""
        if not self.config_path.exists():
            raise ConfigNotFoundError(
                f"配置文件不存在: {self.config_path}"
            )
        if not self.config_path.is_file():
            raise ConfigError(
                f"不是文件: {self.config_path}"
            )
    
    def _parse_config(self):
        """解析配置文件"""
        try:
            with self.config_path.open("r", encoding="utf-8") as f:
                self.config = json.load(f)
        except json.JSONDecodeError as e:
            raise ConfigParseError(
                f"配置文件解析失败: {e}"
            )
        except PermissionError:
            raise ConfigError(
                f"没有权限读取配置文件: {self.config_path}"
            )
    
    def _validate_config(self):
        """验证配置"""
        # 检查必需字段
        for field in self.REQUIRED_FIELDS:
            if field not in self.config:
                raise ConfigValidationError(
                    f"缺少必需字段: {field}"
                )
        
        # 验证日志级别
        log_level = self.config.get("log_level")
        if log_level not in self.VALID_LOG_LEVELS:
            raise ConfigValidationError(
                f"无效的日志级别: {log_level}，"
                f"有效值: {', '.join(self.VALID_LOG_LEVELS)}"
            )
        
        # 验证数据库配置
        db_config = self.config.get("database", {})
        required_db_fields = ["host", "port", "name"]
        for field in required_db_fields:
            if field not in db_config:
                raise ConfigValidationError(
                    f"数据库配置缺少必需字段: {field}"
                )

# 创建示例配置文件
sample_config = {
    "database": {
        "host": "localhost",
        "port": 5432,
        "name": "mydb"
    },
    "server": {
        "host": "0.0.0.0",
        "port": 8080
    },
    "log_level": "INFO"
}

with open("config.json", "w", encoding="utf-8") as f:
    json.dump(sample_config, f, indent=2)

# 使用配置加载器
print("=== 配置文件加载器 ===")

try:
    loader = ConfigLoader("config.json")
    config = loader.load()
    print("配置加载成功!")
    print(f"数据库: {config['database']['host']}:{config['database']['port']}")
    print(f"日志级别: {config['log_level']}")
except ConfigError as e:
    print(f"配置错误: {e}")

7.3 案例3：数据处理管道

# data_pipeline.py

class PipelineError(Exception):
    """管道错误基类"""
    pass

class DataSourceError(PipelineError):
    """数据源错误"""
    pass

class DataProcessingError(PipelineError):
    """数据处理错误"""
    pass

class DataOutputError(PipelineError):
    """数据输出错误"""
    pass

class DataPipeline:
    """数据处理管道"""
    
    def __init__(self):
        self.steps = []
    
    def add_step(self, name, func):
        """添加处理步骤"""
        self.steps.append((name, func))
    
    def process(self, data):
        """处理数据"""
        current_data = data
        
        for step_name, step_func in self.steps:
            try:
                print(f"执行步骤: {step_name}")
                current_data = step_func(current_data)
            except Exception as e:
                raise DataProcessingError(
                    f"步骤 '{step_name}' 执行失败: {e}"
                ) from e
        
        return current_data

# 数据处理函数
def load_data(source):
    """加载数据"""
    if not source:
        raise DataSourceError("数据源不能为空")
    print(f"从 {source} 加载数据")
    return [1, 2, 3, 4, 5]

def clean_data(data):
    """清洗数据"""
    if not data:
        raise DataProcessingError("没有数据可清洗")
    print("清洗数据")
    return [x for x in data if x > 0]

def transform_data(data):
    """转换数据"""
    print("转换数据")
    return [x * 2 for x in data]

def save_data(data, destination):
    """保存数据"""
    if not destination:
        raise DataOutputError("目标位置不能为空")
    print(f"保存数据到 {destination}")
    print(f"数据: {data}")
    return data

# 使用管道
print("=== 数据处理管道 ===")

try:
    pipeline = DataPipeline()
    pipeline.add_step("加载数据", lambda data: load_data("database"))
    pipeline.add_step("清洗数据", clean_data)
    pipeline.add_step("转换数据", transform_data)
    pipeline.add_step("保存数据", lambda data: save_data(data, "output.csv"))
    
    result = pipeline.process(None)
    print("\n处理完成!")
    
except PipelineError as e:
    print(f"\n管道错误: {e}")
    if e.__cause__:
        print(f"原始错误: {e.__cause__}")

8. 自测问题

1. 什么是异常？它和错误有什么区别？
2. try/except/else/finally 的执行顺序是什么？
3. 如何抛出异常？
4. 如何创建自定义异常？
5. 异常处理的最佳实践有哪些？

9. 下集预告

下一集我们将学习Python的模块与包！

参考资料

• Python官方文档: https://docs.python.org/3/tutorial/errors.html
• 《Python编程：从入门到实践》