12、Python 异常处理与调试技巧

Python 异常处理与调试技巧：从防御性编程到生产级排错

本文将系统讲解Python异常处理与调试的核心技术，涵盖从基础语法到生产级调试的完整知识体系，通过20+个代码示例揭示异常处理的深层逻辑，并提供可运行于PyCharm/VSCode的完整代码段。

一、异常处理基础：构建程序防御体系

1.1 try-except-else-finally四联结构

def read_config(filepath):
    try:
        with open(filepath, 'r') as f:
            content = f.read()
    except FileNotFoundError as e:
        print(f"配置文件丢失: {e}")
        content = "default_config"
    except UnicodeDecodeError:
        print("文件编码错误")
        content = ""
    else:
        print("配置文件加载成功")
    finally:
        print("资源清理完成")
        return content

代码解析：

• try块捕获可能异常的IO操作
• 多个except处理不同类型的异常
• else仅在无异常时执行
• finally保证资源释放必定执行

注意事项：

• 避免在finally中使用return语句（可能覆盖异常）
• 异常类型应遵循从具体到宽泛的顺序

1.2 异常对象深度解析

try:
    # 可能抛出ValueError的代码
except ValueError as e:
    print(e.args)    # 原始错误元组
    print(e.__traceback__)  # 堆栈跟踪对象

二、高级异常处理技巧

2.1 自定义异常类设计

class NetworkTimeoutError(Exception):
    """自定义网络超时异常"""
    def __init__(self, url, timeout):
        super().__init__(f"{url}请求超时，阈值{timeout}s")
        self.url = url
        self.timeout = timeout

def fetch_data(url):
    if np.random.rand() > 0.8:
        raise NetworkTimeoutError(url, 5)

2.2 异常链与上下文保留

try:
    conn = database.connect()
except ConnectionError as e:
    raise DataServiceError("数据库连接失败") from e

此时堆栈信息将包含原始异常和新增异常

三、生产级调试技术

3.1 logging模块最佳实践

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

try:
    risky_operation()
except CriticalError:
    logging.exception("关键操作失败")  # 自动记录完整堆栈

3.2 pdb调试器高级用法

断点设置：

import pdb; pdb.set_trace()  # Python 3.7+

或使用breakpoint()（Python 3.7+）

调试命令速查：

• n(ext): 执行下一行
• s(tep): 进入函数内部
• c(ontinue): 继续运行
• l(ist): 显示上下文代码
• p <变量名>: 打印变量值

四、实战：防御性文件处理器开发

class SafeFileProcessor:
    def __init__(self, filename):
        self.filename = filename
        self._validate()

    def _validate(self):
        if not os.path.exists(self.filename):
            raise FileNotFoundError(f"{self.filename} 不存在")
        if not os.access(self.filename, os.R_OK):
            raise PermissionError("文件不可读")

    def process(self):
        try:
            with open(self.filename, 'r') as f:
                self._parse(f.read())
        except UnicodeDecodeError:
            # 尝试不同编码解码
            self._try_alternative_encodings()
        except Exception as e:
            logging.error(f"文件处理失败: {e}")
            raise ProcessingError from e

    # 其他辅助方法省略...

五、练习题与答案代码

练习题列表

1. 实现可重试的网络请求装饰器
2. 创建带错误码的自定义异常体系
3. 使用上下文管理器处理数据库事务
4. 编写自动生成异常日志的装饰器
5. 实现多层级异常传播演示
6. 构建支持断点续传的文件下载器
7. 设计带内存缓存的异常统计器
8. 集成unittest的异常测试用例
9. 实现异步协程的异常处理
10. 开发Flask应用的全局异常处理器

以下是全部10个练习题的完整答案代码及说明：

五、练习题完整答案代码

练习1：可重试网络请求装饰器

import time
import random
from functools import wraps

def retry(max_retries=3, delay=1, exceptions=(Exception,)):
    """通用重试装饰器工厂"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    retries += 1
                    if retries >= max_retries:
                        raise RuntimeError(f"超过最大重试次数 {max_retries}") from e
                    print(f"重试 {retries}/{max_retries}，等待 {delay}s...")
                    time.sleep(delay)
            return func(*args, **kwargs)  # 最后一次尝试
        return wrapper
    return decorator

# 使用示例
@retry(max_retries=5, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
    if random.random() < 0.7:
        raise ConnectionError("模拟连接错误")
    return "数据内容"

练习2：带错误码的自定义异常体系

class AppBaseError(Exception):
    """异常基类"""
    def __init__(self, code, message):
        self.code = code
        self.message = message
        super().__init__(f"[{self.code}] {self.message}")

class DatabaseError(AppBaseError):
    """数据库异常"""
    def __init__(self, message):
        super().__init__(1000, f"数据库错误: {message}")

class AuthError(AppBaseError):
    """认证异常"""
    def __init__(self, message):
        super().__init__(2000, f"认证失败: {message}")

# 使用示例
def login(username):
    if username == "admin":
        raise AuthError("管理员需要二次验证")

练习3：数据库事务上下文管理器

import sqlite3
from contextlib import contextmanager

@contextmanager
def db_transaction(db_path):
    """数据库事务上下文管理器"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()  # 无异常则提交
    except Exception as e:
        conn.rollback()  # 异常回滚
        raise e
    finally:
        conn.close()

# 使用示例
with db_transaction("test.db") as cursor:
    cursor.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))

练习4：自动记录异常的装饰器

import logging
from functools import wraps

logging.basicConfig(level=logging.ERROR)

def log_exceptions(logger=logging.getLogger(__name__)):
    """异常日志记录装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.exception(f"函数 {func.__name__} 执行失败")
                raise
        return wrapper
    return decorator

# 使用示例
@log_exceptions()
def risky_operation():
    raise ValueError("危险操作出错")

练习5：多层级异常传播演示

class Level1Error(Exception): pass
class Level2Error(Exception): pass

def layer1():
    try:
        layer2()
    except Level2Error as e:
        raise Level1Error("Layer1错误") from e

def layer2():
    raise Level2Error("Layer2原始错误")

# 测试代码
try:
    layer1()
except Level1Error as e:
    print(f"捕获到异常: {e}\n原始异常: {e.__cause__}")

练习6：断点续传文件下载器

import os
import requests

def resume_download(url, filename):
    """支持断点续传的文件下载"""
    headers = {}
    if os.path.exists(filename):
        file_size = os.path.getsize(filename)
        headers = {'Range': f'bytes={file_size}-'}
    
    response = requests.get(url, headers=headers, stream=True, timeout=10)
    
    mode = 'ab' if headers else 'wb'
    
    with open(filename, mode) as f:
        for chunk in response.iter_content(chunk_size=1024):
            if chunk:
                f.write(chunk)

# 使用示例
resume_download('https://example.com/largefile.zip', 'largefile.zip')

练习7：异常统计器

from collections import defaultdict

class ExceptionTracker:
    """异常统计单例类"""
    _instance = None
    counts = defaultdict(int)
    
    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def record(self, exc_type):
        self.counts[exc_type.__name__] += 1
    
    def get_report(self):
        return dict(self.counts)

def track_exceptions(func):
    """异常跟踪装饰器"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            tracker = ExceptionTracker()
            tracker.record(type(e))
            raise
    return wrapper

练习8：unittest异常测试用例

import unittest

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

class TestMathOperations(unittest.TestCase):
    def test_divide_normal(self):
        self.assertEqual(divide(10, 2), 5)
    
    def test_divide_by_zero(self):
        with self.assertRaises(ValueError) as context:
            divide(10, 0)
        self.assertEqual(str(context.exception), "除数不能为零")

if __name__ == '__main__':
    unittest.main()

练习9：异步协程异常处理

import asyncio

async def async_task():
    try:
        await asyncio.sleep(1)
        raise RuntimeError("协程内部错误")
    except RuntimeError as e:
        print(f"捕获异步异常: {e}")
        raise  # 重新抛出

async def main():
    try:
        await async_task()
    except Exception as e:
        print(f"外层捕获: {e}")

# Python 3.7+ 运行方式
# asyncio.run(main())

练习10：Flask全局异常处理器

from flask import Flask, jsonify

app = Flask(__name__)

class APIError(Exception):
    def __init__(self, code, message):
        self.code = code
        self.message = message

@app.errorhandler(APIError)
def handle_api_error(error):
    response = jsonify({
        'error_code': error.code,
        'error_message': error.message
    })
    response.status_code = 400
    return response

@app.route('/data')
def get_data():
    raise APIError(1001, "数据获取失败")

if __name__ == '__main__':
    app.run()

代码使用说明

1. 直接运行性：所有代码基于Python 3.7+编写，确保安装必要依赖（requests, flask等）
2. 调试建议：
   • 在PyCharm/VSCode中设置断点逐步执行
   • 使用python -m pytest test_file.py运行单元测试
   • 通过logging.basicConfig(level=logging.DEBUG)查看详细日志
3. 生产环境适配：
   • 异常处理器应记录到文件系统
   • 装饰器类可封装为独立Python模块
   • 数据库连接需替换为真实生产配置

通过实践这些代码，开发者可深入掌握Python异常处理机制，并具备构建健壮生产级应用的能力。建议读者根据实际业务需求调整异常类型和处理逻辑。

六、性能优化与注意事项

1. 异常开销分析：异常处理比条件判断慢3-4倍（基准测试证明）
2. try块作用域优化：尽量缩小try块范围
3. 异常吞噬陷阱：避免裸except导致静默失败
4. 资源泄露防护：结合with语句和finally确保释放

通过本文的学习，开发者将掌握从基础异常处理到生产级调试的完整技能体系，构建出健壮的Python应用程序。文中的代码示例可直接应用于实际项目，建议读者在IDE中逐步调试观察执行流程。