Skip to content

异常处理

异常处理是 Python 程序中处理错误的重要机制。

异常概述

异常层次结构

BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
    ├── StopIteration
    ├── ArithmeticError
    │   ├── ZeroDivisionError
    │   ├── OverflowError
    │   └── FloatingPointError
    ├── LookupError
    │   ├── IndexError
    │   └── KeyError
    ├── TypeError
    ├── ValueError
    ├── FileNotFoundError
    ├── PermissionError
    └── ...

常见异常

python
# TypeError - 类型错误
# "hello" + 1

# ValueError - 值错误
# int("abc")

# IndexError - 索引越界
# [1, 2, 3][10]

# KeyError - 键不存在
# {"a": 1}["b"]

# ZeroDivisionError - 除零
# 1 / 0

# FileNotFoundError - 文件不存在
# open("not_exist.txt")

# AttributeError - 属性不存在
# "hello".not_exist()

# TypeError - 参数类型错误
# len(123)

异常处理

try-except

python
# 基本语法
try:
    result = 10 / 0
except ZeroDivisionError:
    print("除零错误")

# 捕获异常信息
try:
    result = 10 / 0
except ZeroDivisionError as e:
    print(f"错误: {e}")
    print(f"类型: {type(e)}")

# 多个 except
try:
    value = int(input("输入数字: "))
    result = 10 / value
except ValueError:
    print("请输入有效数字")
except ZeroDivisionError:
    print("不能为零")

# 捕获多个异常
try:
    value = int(input("输入数字: "))
    result = 10 / value
except (ValueError, ZeroDivisionError) as e:
    print(f"错误: {e}")

# 捕获所有异常
try:
    # 可能出错的代码
    pass
except Exception as e:
    print(f"发生错误: {e}")

try-except-else-finally

python
try:
    file = open("example.txt", "r")
    content = file.read()
except FileNotFoundError:
    print("文件不存在")
except PermissionError:
    print("没有权限")
else:
    # 没有异常时执行
    print(f"文件内容: {content}")
finally:
    # 无论是否异常都执行
    print("清理资源")
    # file.close()  # 如果文件打开成功

# 使用 with 语句更简洁
try:
    with open("example.txt", "r") as f:
        content = f.read()
except FileNotFoundError:
    print("文件不存在")

异常链

python
def process_data(data):
    try:
        return int(data)
    except ValueError as e:
        raise ValueError(f"无法处理数据: {data}") from e

try:
    process_data("abc")
except ValueError as e:
    print(f"错误: {e}")
    print(f"原因: {e.__cause__}")

抛出异常

raise 语句

python
# 抛出异常
def set_age(age):
    if age < 0 or age > 150:
        raise ValueError("年龄必须在 0-150 之间")
    return age

try:
    set_age(-5)
except ValueError as e:
    print(e)

# 重新抛出异常
def process():
    try:
        # 一些操作
        pass
    except Exception as e:
        print("记录日志")
        raise  # 重新抛出

# 抛出自定义异常
class CustomError(Exception):
    pass

raise CustomError("自定义错误信息")

assert 语句

python
def calculate_average(numbers):
    assert len(numbers) > 0, "列表不能为空"
    return sum(numbers) / len(numbers)

try:
    calculate_average([])
except AssertionError as e:
    print(e)

# 禁用断言: python -O script.py

自定义异常

定义异常类

python
class InsufficientFundsError(Exception):
    """余额不足异常"""
    
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        super().__init__(
            f"余额不足: 当前余额 {balance}, 需要 {amount}"
        )

class BankAccount:
    def __init__(self, balance):
        self.balance = balance
    
    def withdraw(self, amount):
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)
        self.balance -= amount
        return self.balance

# 使用
account = BankAccount(100)
try:
    account.withdraw(150)
except InsufficientFundsError as e:
    print(e)
    print(f"当前余额: {e.balance}")

异常层次结构

python
class AppError(Exception):
    """应用程序基础异常"""
    pass

class DatabaseError(AppError):
    """数据库异常"""
    pass

class ConnectionError(DatabaseError):
    """连接异常"""
    pass

class QueryError(DatabaseError):
    """查询异常"""
    pass

# 使用
def query_database(sql):
    if not sql:
        raise QueryError("SQL 不能为空")
    # 执行查询

try:
    query_database("")
except DatabaseError as e:
    print(f"数据库错误: {e}")
except AppError as e:
    print(f"应用错误: {e}")

异常处理最佳实践

1. 只捕获能处理的异常

python
# 不推荐
try:
    # 很多操作
    pass
except:
    pass  # 吞掉所有异常

# 推荐
try:
    value = int(input("输入数字: "))
except ValueError:
    print("请输入有效数字")

2. 使用具体的异常类型

python
# 不推荐
try:
    file = open("data.txt")
except Exception:
    print("错误")

# 推荐
try:
    file = open("data.txt")
except FileNotFoundError:
    print("文件不存在")
except PermissionError:
    print("没有权限")

3. 不要使用异常控制流程

python
# 不推荐
try:
    value = my_dict[key]
except KeyError:
    value = None

# 推荐
value = my_dict.get(key)

4. 提供有意义的异常信息

python
# 不推荐
raise ValueError("无效输入")

# 推荐
raise ValueError(f"年龄必须为正整数,当前值: {age}")

5. 使用上下文管理器

python
# 不推荐
file = None
try:
    file = open("data.txt")
    # 操作
finally:
    if file:
        file.close()

# 推荐
with open("data.txt") as file:
    # 操作
    pass

上下文管理器

自定义上下文管理器

python
class Timer:
    def __enter__(self):
        import time
        self.start = time.time()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.end = time.time()
        print(f"耗时: {self.end - self.start:.2f}秒")
        return False  # 不抑制异常

with Timer():
    # 执行代码
    sum(range(1000000))

# 使用 contextlib
from contextlib import contextmanager

@contextmanager
def timer():
    import time
    start = time.time()
    yield
    end = time.time()
    print(f"耗时: {end - start:.2f}秒")

with timer():
    sum(range(1000000))

异常处理的上下文管理器

python
from contextlib import suppress

# 抑制特定异常
with suppress(FileNotFoundError):
    os.remove("not_exist.txt")

# 等价于
try:
    os.remove("not_exist.txt")
except FileNotFoundError:
    pass

日志记录

使用 logging 模块

python
import logging

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='app.log'
)

logger = logging.getLogger(__name__)

def divide(a, b):
    try:
        result = a / b
        logger.info(f"除法结果: {a} / {b} = {result}")
        return result
    except ZeroDivisionError:
        logger.error("除零错误", exc_info=True)
        raise

try:
    divide(10, 0)
except ZeroDivisionError:
    pass

实践示例

输入验证

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

def validate_email(email):
    """验证邮箱格式"""
    if not email or "@" not in email:
        raise ValidationError("邮箱格式不正确")
    return email

def validate_age(age):
    """验证年龄"""
    if not isinstance(age, int):
        raise ValidationError("年龄必须是整数")
    if age < 0 or age > 150:
        raise ValidationError("年龄必须在 0-150 之间")
    return age

def validate_user(data):
    """验证用户数据"""
    errors = []
    
    try:
        validate_email(data.get("email"))
    except ValidationError as e:
        errors.append(("email", str(e)))
    
    try:
        validate_age(data.get("age"))
    except ValidationError as e:
        errors.append(("age", str(e)))
    
    if errors:
        raise ValidationError(f"验证失败: {errors}")
    
    return True

# 使用
try:
    validate_user({"email": "invalid", "age": -5})
except ValidationError as e:
    print(e)

重试机制

python
import time
from functools import wraps

def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        print(f"第 {attempt + 1} 次尝试失败: {e}")
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2, exceptions=(ConnectionError,))
def fetch_data(url):
    # 模拟网络请求
    import random
    if random.random() < 0.7:
        raise ConnectionError("连接失败")
    return "数据"

try:
    data = fetch_data("http://example.com")
    print(data)
except ConnectionError as e:
    print(f"最终失败: {e}")

资源管理

python
from contextlib import contextmanager

@contextmanager
def managed_file(filename, mode):
    """安全的文件操作"""
    f = None
    try:
        f = open(filename, mode)
        yield f
    except IOError as e:
        print(f"文件操作错误: {e}")
        raise
    finally:
        if f:
            f.close()

with managed_file("test.txt", "w") as f:
    f.write("Hello")

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print("建立连接")
        self.connection = {"connected": True}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f"发生错误: {exc_val}")
            print("回滚事务")
        else:
            print("提交事务")
        print("关闭连接")
        return False

with DatabaseConnection("mysql://...") as conn:
    # 执行数据库操作
    pass