前言
异常处理是区分”能用”与”用好”Python 的分水岭。优秀的异常处理能让程序在出错时优雅降级、留下可追溯的日志；而糟糕的异常处理则会让 bug 藏匿于无声的失败中，排查成本成倍增加。
本文将系统讲解：
- 异常机制的底层原理与语法细节
- 自定义异常的设计模式
- 调试工具的实战用法（pdb、logging、traceback）
- 反模式警示与最佳实践
- 常见场景的代码模板
异常基础：try/except/finally
基本语法结构
# 基础结构
try:
# 可能抛出异常的代码
result = 10 / 0
except ZeroDivisionError:
# 处理特定异常
print("除数不能为零")
except (ValueError, TypeError) as e:
# 同时捕获多种异常
print(f"值错误: {e}")
except Exception as e:
# 兜底捕获所有异常
print(f"未知错误: {e}")
else:
# 仅在 try 块成功时执行（可选）
print("计算成功")
finally:
# 无论是否异常都执行（可选）
print("清理资源")异常捕获的执行顺序
# 异常捕获顺序很重要！
try:
raise ValueError("原始错误")
except ValueError as e:
print(f"捕获 ValueError: {e}") # 先匹配这个
except Exception as e:
print(f"捕获 Exception: {e}") # 永远不会执行原则：子异常类必须写在父异常类前面
# ❌ 错误示范
try:
risky_operation()
except Exception: # 永远先匹配这个
pass
except ValueError: # 永远不会执行
pass
# ✅ 正确写法
try:
risky_operation()
except ValueError: # 先匹配具体的
pass
except Exception: # 再匹配通用的
pass异常对象的核心属性
try:
1 / 0
except ZeroDivisionError as e:
print(f"异常类型: {type(e).__name__}") # ZeroDivisionError
print(f"错误信息: {e}") # division by zero
print(f"完整追踪:\n{traceback.format_exc()}") # 堆栈追踪
# 异常对象的常用属性
print(f"异常发生位置: {e.__traceback__.tb_frame.f_code.co_filename}")
print(f"异常函数名: {e.__traceback__.tb_frame.f_code.co_name}")
print(f"行号: {e.__traceback__.tb_lineno}")重新抛出异常
# 方法1：直接 raise（保持原始异常）
try:
some_operation()
except SomeError:
print("记录日志")
raise # 重新抛出，不带参数
# 方法2：raise from（显式异常链）
try:
db.connect()
except ConnectionError as e:
raise ValueError("无法连接到数据库") from e # 新异常源于原异常
# 方法3：raise from None（隐藏异常链）
try:
legacy_code()
except Exception:
raise NewError("新错误") from None # 隐藏原始异常异常层级与自定义异常
Python 内置异常层级
BaseException
├── SystemExit
├── KeyboardInterrupt
└── Exception
├── StopIteration
├── ArithmeticError
│ ├── FloatingPointError
│ ├── OverflowError
│ └── ZeroDivisionError
├── LookupError
│ ├── IndexError
│ └── KeyError
├── OSError (IOError)
│ ├── FileNotFoundError
│ ├── PermissionError
│ └── TimeoutError
├── ValueError
├── TypeError
└── ...自定义异常设计原则
# 原则1：继承合适的基类
class AppError(Exception):
"""应用层异常基类"""
pass
class ValidationError(AppError):
"""数据验证错误"""
pass
class AuthenticationError(AppError):
"""认证错误"""
pass
class NetworkError(AppError):
"""网络相关错误"""
pass
# 原则2：添加有意义的属性
class APIError(AppError):
"""API 调用错误"""
def __init__(self, code, message, response=None):
self.code = code
self.message = message
self.response = response
super().__init__(f"[{code}] {message}")
# 使用
raise APIError(404, "资源不存在", response={"id": 123})异常工厂模式
class DatabaseError(Exception):
"""数据库错误工厂"""
@staticmethod
def connection_failed(host, reason):
return DatabaseError(f"连接数据库 {host} 失败: {reason}")
@staticmethod
def query_failed(sql, reason):
return DatabaseError(f"执行 SQL 失败: {sql[:50]}... 原因: {reason}")
@staticmethod
def timeout(seconds):
return DatabaseError(f"查询超时 ({seconds}s)")
# 使用
try:
db.query(sql)
except Exception as e:
raise DatabaseError.query_failed(sql, str(e))上下文管理器与资源清理
with 语句原理
# 上下文管理器协议
class MyResource:
def __enter__(self):
"""进入 with 块时调用"""
print("获取资源")
return self # as 后的变量接收此值
def __exit__(self, exc_type, exc_val, exc_tb):
"""离开 with 块时调用"""
if exc_type:
print(f"异常类型: {exc_type.__name__}")
print(f"异常值: {exc_val}")
# 返回 True 抑制异常，False 或 None 传播异常
return False
print("释放资源")
return False
# 使用
with MyResource() as r:
print("使用资源中...")
# 如果这里抛异常，__exit__ 会收到异常信息@contextmanager 装饰器
from contextlib import contextmanager
@contextmanager
def managed_resource(name):
"""简化版上下文管理器"""
print(f"获取 {name}")
try:
yield name # 相当于 __enter__ 返回值
finally:
print(f"释放 {name}")
with managed_resource("数据库连接") as conn:
print(f"使用 {conn}")
# 输出:
# 获取 数据库连接
# 使用 数据库连接
# 释放 数据库连接嵌套上下文与组合
from contextlib import ExitStack
# 场景：需要同时管理多个资源
def process_files(files):
"""同时打开多个文件"""
with ExitStack() as stack:
file_handles = [
stack.enter_context(open(f, 'r'))
for f in files
]
# 处理文件
for f in file_handles:
process(f)
# 所有文件自动关闭
# 场景：条件性资源清理
@contextmanager
def temporary_file(path):
"""临时文件，使用后自动删除"""
try:
yield Path(path)
finally:
if Path(path).exists():
Path(path).unlink()异常与 finally 的陷阱
# 陷阱1：finally 中 return 会覆盖 try 中的异常
def bad_example():
try:
raise ValueError("原始异常")
finally:
return 42 # 异常被吞掉，函数返回 42
print(bad_example()) # 输出 42，不见异常
# 陷阱2：finally 先于 except 中的 return 执行
def flow_demo():
try:
print("1 - try 开始")
raise ValueError()
except:
print("2 - except 捕获")
return # return 延迟到 finally 后执行
finally:
print("3 - finally 执行")
# 输出顺序: 1 -> 2 -> 3
# 陷阱3：finally 中的异常会覆盖原异常
def掩盖异常():
try:
raise ValueError("原异常")
finally:
raise RuntimeError("finally 中的异常") # 原异常被掩盖
try:
掩盖异常()
except RuntimeError as e:
print(e) # 只看到 "finally 中的异常"调试工具全景图
1. 内置 traceback 模块
import traceback
import sys
# 格式化异常信息
def log_exception():
exc_type, exc_value, exc_tb = sys.exc_info()
print("=== 异常摘要 ===")
print(f"类型: {exc_type.__name__}")
print(f"信息: {exc_value}")
print("\n=== 完整堆栈 ===")
traceback.print_tb(exc_tb)
# 获取字符串格式
stack_str = "".join(traceback.format_tb(exc_tb))
print(f"\n堆栈字符串:\n{stack_str}")
# 使用
try:
[1, 2, 3][10] # 触发 IndexError
except:
log_exception()2. 高级 traceback 格式化
import traceback
import sys
from pathlib import Path
def print_exception_details(exc_type, exc_value, exc_tb, max_lines=20):
"""美化异常输出"""
if not exc_tb:
return
print("═" * 60)
print(f"🔴 {exc_type.__name__}: {exc_value}")
print("═" * 60)
# 逐帧显示
tb = exc_tb
frames = []
while tb:
frame = tb.tb_frame
code = frame.f_code
frames.append({
"file": Path(code.co_filename).name,
"func": code.co_name,
"line": tb.tb_lineno,
"locals": {k: repr(v)[:50]
for k, v in frame.f_locals.items()
if not k.startswith('_')}
})
tb = tb.tb_next
# 过滤到最近几帧
for i, frame in enumerate(reversed(frames[-max_lines:])):
print(f"\n📍 Frame {len(frames) - i - 1}")
print(f" 文件: {frame['file']}")
print(f" 函数: {frame['func']}()")
print(f" 行号: {frame['line']}")
if frame['locals']:
print(f" 局部变量:")
for k, v in list(frame['locals'].items())[:5]:
print(f" {k} = {v}")
print("\n" + "═" * 60)
# 使用
try:
def inner():
x = 42
y = [1, 2]
return y[x]
inner()
except:
print_exception_details(*sys.exc_info())3. pdb 调试器实战
# 方法1：代码中设置断点
import pdb
def buggy_function(data):
pdb.set_trace() # 程序在此暂停，进入交互式调试器
result = []
for i, item in enumerate(data):
# 在调试器中可以：
# - n (next): 执行下一行
# - s (step): 进入函数
# - p variable: 打印变量
# - l (list): 查看当前代码
# - c (continue): 继续执行
# - q (quit): 退出调试
result.append(item * 2)
return result
buggy_function([1, 2, 3])
# 方法2：命令行启动
# python -m pdb script.py
# 方法3：事后调试
import pdb
import traceback
try:
risky_code()
except Exception:
pdb.pm() # 事后进入调试器，查看异常发生时的状态4. breakpoint() 内置函数（Python 3.7+）
# Python 3.7+ 的 breakpoint() 更强大
breakpoint() # 等价于 pdb.set_trace()
# 配置使用其他调试器
import debugpy
breakpoint() # 自动使用 debugpy
# 或在环境中设置
# export PYTHONBREAKPOINT=ipdb.set_trace5. logging 模块高级用法
import logging
from pathlib import Path
# 配置日志系统
def setup_logger(name, log_file=None, level=logging.DEBUG):
logger = logging.getLogger(name)
logger.setLevel(level)
formatter = logging.Formatter(
'%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 控制台输出
console = logging.StreamHandler()
console.setFormatter(formatter)
logger.addHandler(console)
# 文件输出（带轮转）
if log_file:
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
# 使用
logger = setup_logger(__name__, 'app.log')
try:
result = calculate()
logger.info(f"计算成功: {result}")
except ValueError as e:
logger.error(f"验证错误: {e}", exc_info=True) # exc_info=True 打印堆栈
except Exception as e:
logger.critical(f"系统错误: {e}", exc_info=True)
raise
# 条件日志（避免昂贵操作的日志开销）
if logger.isEnabledFor(logging.DEBUG):
logger.debug(f"详细数据: {expensive_to_serialize()}")实战调试技巧
技巧1：断言的艺术
# 断言用于开发期捕获"不可能发生"的错误
# 生产环境可用 -O 标志禁用
def calculate_discount(price, discount):
assert 0 <= discount <= 1, f"折扣率必须在 0-1 之间，当前值: {discount}"
assert price >= 0, f"价格不能为负: {price}"
final_price = price * (1 - discount)
assert final_price >= 0, "计算结果异常"
return final_price
# 使用自定义 AssertionError
class PriceError(AssertionError):
"""价格计算错误"""
pass
def safe_divide(a, b):
if b == 0:
raise PriceError(f"除数不能为零: a={a}, b={b}")
return a / b技巧2：异常链与日志追踪
import logging
import traceback
logger = logging.getLogger(__name__)
def process_data(raw_data):
"""异常处理与日志记录的标准模式"""
try:
# 解析数据
data = parse(raw_data)
# 验证数据
validate(data)
# 处理数据
return transform(data)
except ValidationError as e:
logger.warning(f"数据验证失败: {e}")
raise DataError(f"无效数据: {e}") from e
except Exception as e:
logger.error(
f"处理数据时发生未知错误: {e}",
exc_info=True, # 打印完整堆栈
extra={"raw_data": str(raw_data)[:100]} # 添加上下文
)
raise ProcessingError("数据处理失败") from e技巧3：使用警告系统
import warnings
# 警告不会终止程序，但能引起注意
warnings.warn("这个 API 已弃用，请使用新方法", DeprecationWarning)
warnings.warn("性能可能受影响", PerformanceWarning)
# 警告过滤器配置
warnings.filterwarnings(
"error", # 将警告转为异常
category=DeprecationWarning
)
# 按模块过滤
warnings.filterwarnings(
"ignore",
message=".*deprecated.*",
category=PendingDeprecationWarning
)技巧4：数据校验模式
from typing import Any, Callable
def validate(value: Any, *rules: Callable) -> Any:
"""链式数据校验"""
for rule in rules:
if not rule(value):
raise ValidationError(f"值 {value!r} 不满足规则: {rule.__name__}")
return value
def is_positive(x): return x > 0
def is_integer(x): return isinstance(x, int)
def in_range(min_v, max_v):
def _check(x): return min_v <= x <= max_v
return _check
# 使用
try:
age = validate(
user_input,
is_integer,
is_positive,
in_range(0, 150)
)
except ValidationError as e:
print(f"校验失败: {e}")技巧5：异常处理装饰器
import functools
import time
import logging
logger = logging.getLogger(__name__)
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
"""重试装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts - 1:
raise
logger.warning(
f"{func.__name__} 第 {attempt + 1} 次失败: {e}，"
f"{delay}s 后重试..."
)
time.sleep(delay)
return wrapper
return decorator
def handle_errors(default=None, log=True):
"""异常处理装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
if log:
logger.error(f"{func.__name__} 执行失败: {e}")
if callable(default):
return default(e)
return default
return wrapper
return decorator
# 使用
@retry(max_attempts=3, delay=2, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
return requests.get(url)
@handle_errors(default={}, log=True)
def parse_json(data):
return json.loads(data)技巧6：IPython/jupyter 调试增强
# 在 IPython 或 Jupyter 中使用
%debug # 事后调试最近的异常
%timeit # 测试代码执行时间
# 变量检查
# %who - 列出所有变量
# %whos - 列出所有变量及类型
# %pdb - 自动启用 pdb常见反模式与最佳实践
反模式警示
| 反模式 | 问题 | 正确做法 |
|---|---|---|
except: pass | 吞掉所有异常 | except SomeError: handle_it() |
裸 except Exception | 过于宽泛 | 具体异常类型 |
raise Exception(e) | 丢失原始堆栈 | raise 不带参数重新抛出 |
| 异常用于流程控制 | 可读性差 | 使用 if/else |
| 异常中执行 IO | 性能问题 | 异常应轻量 |
异常处理最佳实践清单
# ✅ 1. 具体异常优先
try:
config = json.load(f)
except json.JSONDecodeError as e:
logger.error(f"配置文件格式错误: {e}")
# ❌ 2. 避免过于宽泛的捕获
try:
config = json.load(f)
except Exception: # 捕获了文件不存在、权限问题等
pass
# ✅ 3. 异常包含上下文
class OrderNotFoundError(Exception):
def __init__(self, order_id):
self.order_id = order_id
super().__init__(f"订单不存在: {order_id}")
raise OrderNotFoundError(order_id)
# ❌ 4. 避免异常中的副作用
try:
process(data)
except Exception:
cleanup() # 异常处理函数不应抛异常
raise
# ✅ 5. EAFP vs LBYL 风格选择
# EAFP (Easier to Ask for Forgiveness than Permission)
try:
value = data[key]
except KeyError:
value = default
# LBYL (Look Before You Leap)
if key in data:
value = data[key]
else:
value = default
# 建议：字典/列表用 EAFP，文件/网络用 LBYL（性能更好）分层异常处理
# 层级1：最外层 - 统一错误响应（API/Web）
@app.errorhandler(ValidationError)
def handle_validation(e):
return {"error": str(e)}, 400
# 层级2：业务逻辑层 - 异常转换
class OrderService:
def create_order(self, data):
try:
self.validate(data)
return self.repository.save(data)
except DBError as e:
raise OrderError(f"创建订单失败: {e}") from e
# 层级3：数据访问层 - 具体异常
class OrderRepository:
def save(self, data):
try:
return self.db.insert(data)
except IntegrityError:
raise DBError("数据完整性冲突")排错指南
常见错误与解决方案
错误1：异常被意外吞掉
# 问题
try:
risky_operation()
except:
pass # 所有异常都消失得无影无踪
# 诊断：添加日志
import logging
logger = logging.getLogger(__name__)
try:
risky_operation()
except Exception as e:
logger.exception("操作失败") # 自动包含堆栈
raise错误2：异常信息不完整
# 问题：只有错误类型，没有上下文
try:
process(user_id=123)
except UserNotFoundError:
raise ValueError("用户不存在") # 丢失 user_id 信息
# 解决：包含所有上下文
try:
process(user_id=123)
except UserNotFoundError:
raise ValueError(f"用户不存在: user_id={123}") from None错误3：finally 中的异常掩盖原异常
# 问题
try:
raise OriginalError("原异常")
finally:
cleanup() # 如果 cleanup 抛异常，原异常丢失
# 解决：显式处理
try:
raise OriginalError("原异常")
except OriginalError:
try:
cleanup()
except:
pass # 记录但不传播 cleanup 的异常
raise # 重新抛出原异常错误4：异步代码中的异常
import asyncio
async def bad_async():
try:
await risky_async_call()
except Exception:
# 异常在异步任务中传播
pass # 异常被吞掉！
# 正确：确保异常被记录
async def good_async():
try:
await risky_async_call()
except Exception as e:
logger.exception(f"异步任务失败: {e}")
raise错误5：循环中的异常处理
# 问题
for item in items:
try:
process(item)
except Exception:
pass # 一个失败就静默跳过
# 解决：记录所有错误
errors = []
for item in items:
try:
process(item)
except Exception as e:
errors.append((item, e))
if errors:
logger.error(f"处理失败 {len(errors)} 项:")
for item, e in errors:
logger.error(f" {item}: {e}")调试工作流
# 1. 复现问题
# 确保能稳定触发问题
# 2. 隔离问题
def isolate_issue():
# 二分法逐步缩小范围
pass
# 3. 添加诊断日志
import logging
logging.basicConfig(level=logging.DEBUG)
# 4. 使用断点
import pdb; pdb.set_trace()
# 5. 编写最小复现用例
def test_reproduction():
"""最小化复现代码"""
raise AssertionError("期望与实际不符")
# 6. 修复并验证
# 7. 添加回归测试
import pytest
def test_that_fails_then_passes():
"""捕获之前失败的场景"""
with pytest.raises(CustomError):
risky_function()总结
核心要点
- 异常捕获要具体：优先捕获最具体的异常类型
- 保留异常链：使用
raise ... from e保持上下文 - 上下文管理器：
with语句确保资源正确释放 - 日志记录：
logger.exception()自动包含堆栈信息 - 防御性编程：结合断言与异常处理
一图总结
┌─────────────────────────────────────────────────────────────┐
│ 异常处理速查表 │
├──────────────────┬──────────────────────────────────────────┤
│ 场景 │ 做法 │
├──────────────────┼──────────────────────────────────────────┤
│ 捕获多种异常 │ except (A, B) as e: │
│ 重新抛出异常 │ raise # 不带参数 │
│ 异常链 │ raise NewError() from e │
│ 抑制异常链 │ raise NewError() from None │
│ 记录并重新抛出 │ logger.exception(); raise │
│ 资源清理 │ with resource: ... # 自动清理 │
│ 重试机制 │ @retry(max_attempts=3) │
│ 事后调试 │ pdb.pm() / %debug (IPython) │
└──────────────────┴──────────────────────────────────────────┘推荐工具链
| 场景 | 工具 | 说明 |
|---|---|---|
| 交互式调试 | pdb / ipdb | 命令行调试器 |
| 图形调试 | IDE 内置 | VS Code / PyCharm |
| 日志 | logging | 标准库，推荐配置后使用 |
| 性能分析 | cProfile | 内置性能分析器 |
| 单元测试 | pytest | 完善的断言与异常测试支持 |
| 静态检查 | mypy / pyright | 类型检查可发现潜在错误 |
💡 提示: 生产环境中，建议使用 Sentry、Bugsnag 等 APM 工具收集异常信息。 📚 扩展阅读:
原创内容，版权所有。未经授权，禁止转载。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容