10 - 错误处理
第 10 章:错误处理
掌握 Python 的异常处理机制,编写健壮、可维护的代码。
10.1 异常基础
10.1.1 常见异常类型
| 异常 | 说明 | 示例 |
|---|---|---|
ValueError | 值不正确 | int("abc") |
TypeError | 类型不匹配 | "1" + 2 |
KeyError | 字典键不存在 | {}["key"] |
IndexError | 索引越界 | [][0] |
AttributeError | 属性不存在 | None.x |
FileNotFoundError | 文件不存在 | open("不存在.txt") |
ImportError | 导入失败 | import 不存在的模块 |
ZeroDivisionError | 除以零 | 1 / 0 |
RuntimeError | 运行时错误 | 一般性运行错误 |
StopIteration | 迭代器耗尽 | next(iter([])) |
10.1.2 异常层次
BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
├── ValueError
├── TypeError
├── KeyError
├── IndexError
├── AttributeError
├── OSError
│ ├── FileNotFoundError
│ ├── PermissionError
│ └── FileExistsError
├── RuntimeError
├── StopIteration
├── ArithmeticError
│ ├── ZeroDivisionError
│ └── OverflowError
└── ...
10.2 try-except 语句
10.2.1 基本语法
try:
result = 10 / int(input("输入一个数字: "))
print(f"结果: {result}")
except ValueError:
print("请输入有效数字!")
except ZeroDivisionError:
print("不能除以零!")
10.2.2 捕获多个异常
# 方式一:分别处理
try:
data = json.loads(user_input)
except json.JSONDecodeError:
print("JSON 格式错误")
except ValueError:
print("值错误")
# 方式二:合并处理
try:
process(data)
except (ValueError, TypeError) as e:
print(f"数据错误: {e}")
10.2.3 获取异常信息
try:
result = 10 / 0
except ZeroDivisionError as e:
print(f"异常类型: {type(e).__name__}") # ZeroDivisionError
print(f"异常消息: {e}") # division by zero
print(f"异常参数: {e.args}") # ('division by zero',)
10.2.4 完整的 try 语句
try:
result = risky_operation()
except ValueError as e:
handle_value_error(e)
except (TypeError, RuntimeError) as e:
handle_other_error(e)
except Exception as e:
handle_unexpected_error(e)
else:
# 没有异常时执行
print(f"成功: {result}")
finally:
# 无论如何都执行(清理资源)
cleanup()
10.2.5 异常链
# 原始异常被保留(Python 3)
try:
try:
result = int("abc")
except ValueError as original:
raise RuntimeError("数据转换失败") from original
except RuntimeError as e:
print(f"错误: {e}")
print(f"原因: {e.__cause__}")
# 抑制异常链
try:
result = int("abc")
except ValueError:
raise RuntimeError("数据转换失败") from None # 不显示原始异常
10.3 自定义异常
10.3.1 定义异常层次
class AppError(Exception):
"""应用基础异常。"""
pass
class ValidationError(AppError):
"""数据验证异常。"""
def __init__(self, field: str, message: str):
self.field = field
self.message = message
super().__init__(f"验证失败 - {field}: {message}")
class NotFoundError(AppError):
"""资源不存在异常。"""
def __init__(self, resource: str, identifier: str):
self.resource = resource
self.identifier = identifier
super().__init__(f"{resource} '{identifier}' 不存在")
class PermissionError(AppError):
"""权限不足异常。"""
pass
# 使用
def get_user(user_id: int) -> dict:
if user_id <= 0:
raise ValidationError("user_id", "必须为正整数")
users = {1: {"name": "Alice"}, 2: {"name": "Bob"}}
if user_id not in users:
raise NotFoundError("用户", str(user_id))
return users[user_id]
try:
user = get_user(99)
except NotFoundError as e:
print(e) # 用户 '99' 不存在
print(e.resource) # 用户
print(e.identifier) # 99
except ValidationError as e:
print(e) # 验证失败 - user_id: 必须为正整数
10.3.2 带错误码的异常
from enum import IntEnum
class ErrorCode(IntEnum):
UNKNOWN = 1000
VALIDATION_ERROR = 1001
NOT_FOUND = 1002
PERMISSION_DENIED = 1003
RATE_LIMITED = 1004
class APIError(Exception):
def __init__(self, code: ErrorCode, message: str, details: dict | None = None):
self.code = code
self.message = message
self.details = details or {}
super().__init__(message)
def to_dict(self) -> dict:
return {
"error": {
"code": self.code.value,
"message": self.message,
"details": self.details,
}
}
# 使用
raise APIError(
ErrorCode.VALIDATION_ERROR,
"请求参数无效",
{"field": "email", "reason": "格式不正确"}
)
10.4 上下文管理器
10.4.1 with 语句
# 文件操作
with open("data.txt") as f:
content = f.read()
# 文件自动关闭
# 多个上下文管理器
with open("input.txt") as fin, open("output.txt", "w") as fout:
fout.write(fin.read())
10.4.2 自定义上下文管理器
class DatabaseConnection:
"""数据库连接上下文管理器。"""
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.connection = None
def __enter__(self):
print(f"连接数据库: {self.connection_string}")
self.connection = {"connected": True, "string": self.connection_string}
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"断开数据库连接")
self.connection = None
# 返回 True 会抑制异常,False/None 不抑制
return False
with DatabaseConnection("postgresql://localhost/mydb") as conn:
print(f"查询中... 连接状态: {conn['connected']}")
# 断开数据库连接(即使发生异常也会执行)
10.4.3 contextlib 模块
from contextlib import contextmanager, suppress, redirect_stdout
import io
# @contextmanager 装饰器
@contextmanager
def timer(label: str):
import time
start = time.perf_counter()
try:
yield # yield 的值会赋给 as 变量
finally:
elapsed = time.perf_counter() - start
print(f"{label}: {elapsed:.4f} 秒")
with timer("数据处理"):
time.sleep(0.5)
# suppress:忽略特定异常
with suppress(FileNotFoundError):
open("不存在的文件.txt")
# 重定向输出
buffer = io.StringIO()
with redirect_stdout(buffer):
print("这会写入 buffer")
output = buffer.getvalue()
10.5 断言(assert)
# 语法:assert condition, message
def calculate_discount(price: float, discount: float) -> float:
assert 0 <= discount <= 1, f"折扣必须在 0-1 之间,收到 {discount}"
assert price >= 0, f"价格不能为负,收到 {price}"
return price * (1 - discount)
print(calculate_discount(100, 0.2)) # 80.0
# calculate_discount(100, 1.5) # AssertionError: 折扣必须在 0-1 之间
# ⚠️ assert 可能被 -O 优化标志禁用
# 生产环境使用显式检查
| 用途 | 推荐方式 |
|---|---|
| 开发时检查不变量 | assert |
| 验证用户输入 | if + raise |
| 检查函数前置条件 | if + raise |
| 检查后置条件 | assert 或 if + raise |
10.6 日志
10.6.1 基本使用
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
# 使用
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告")
logger.error("错误")
logger.critical("严重错误")
# 带异常信息
try:
result = 1 / 0
except ZeroDivisionError:
logger.exception("计算错误") # 自动记录堆栈信息
10.6.2 日志级别
| 级别 | 数值 | 使用场景 |
|---|---|---|
| DEBUG | 10 | 调试细节 |
| INFO | 20 | 一般操作记录 |
| WARNING | 30 | 警告(可恢复) |
| ERROR | 40 | 错误(功能受损) |
| CRITICAL | 50 | 严重错误(系统崩溃) |
10.6.3 配置日志处理器
import logging
from logging.handlers import RotatingFileHandler
# 创建 logger
logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)
# 控制台处理器
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
# 文件处理器(轮转)
file_handler = RotatingFileHandler(
"app.log",
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5,
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
))
logger.addHandler(console)
logger.addHandler(file_handler)
10.7 最佳实践
10.7.1 异常处理原则
# ✅ 精确捕获
try:
value = my_dict[key]
except KeyError:
value = default
# ❌ 过于宽泛
try:
value = my_dict[key]
except Exception:
value = default
# ✅ 保持 try 块小
# ❌ 把整个函数放在 try 里
try:
data = parse(input)
result = process(data)
save(result)
except ...:
...
10.7.2 EAFP 原则
# EAFP: Easier to Ask Forgiveness than Permission(先做再说)
# LBYL: Look Before You Leap(三思后行)
if key in my_dict:
value = my_dict[key]
else:
value = default
# EAFP(Pythonic)
try:
value = my_dict[key]
except KeyError:
value = default
# 更好:使用 get()
value = my_dict.get(key, default)
10.7.3 常见模式
# 重试装饰器
import time
from functools import wraps
def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts:
raise
logger.warning(f"尝试 {attempt} 失败: {e}")
time.sleep(current_delay)
current_delay *= backoff
return wrapper
return decorator
@retry(max_attempts=3, delay=0.5)
def fetch_data(url: str) -> dict:
import requests
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
10.8 注意事项
🔴 注意:
- 不要捕获
BaseException或裸except:(会吞掉KeyboardInterrupt) finally中不要使用return(会覆盖try/except的返回值)assert会被-O优化模式禁用,不要用于验证用户输入- 日志不要记录敏感信息(密码、token 等)
💡 提示:
- 异常也是 API 的一部分,文档中要说明可能抛出的异常
- 使用
logging而非print记录日志 - 自定义异常继承
Exception,不要继承BaseException - 使用
contextlib.suppress替代空的try/except
📌 业务场景:
import logging
from contextlib import contextmanager
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class OrderResult:
order_id: str
status: str
message: str
class OrderError(Exception):
pass
class InsufficientStockError(OrderError):
def __init__(self, product: str, requested: int, available: int):
self.product = product
self.requested = requested
self.available = available
super().__init__(f"{product} 库存不足: 需要 {requested},可用 {available}")
@contextmanager
def order_transaction(order_id: str):
"""订单事务上下文管理器。"""
logger.info(f"开始订单事务: {order_id}")
try:
yield
logger.info(f"订单事务完成: {order_id}")
except OrderError as e:
logger.error(f"订单失败: {order_id} - {e}")
raise
except Exception as e:
logger.critical(f"订单异常: {order_id} - {e}", exc_info=True)
raise OrderError(f"订单处理异常: {e}") from e
def process_order(order_id: str, product: str, quantity: int, stock: int) -> OrderResult:
with order_transaction(order_id):
if quantity <= 0:
raise ValueError("数量必须为正数")
if quantity > stock:
raise InsufficientStockError(product, quantity, stock)
# 模拟订单处理
logger.info(f"扣减库存: {product} x{quantity}")
return OrderResult(
order_id=order_id,
status="success",
message=f"订单 {order_id} 创建成功",
)
10.9 扩展阅读
- Errors and Exceptions - Python 教程
- Built-in Exceptions
- logging 模块文档
- contextlib 模块文档
- 《Python Crash Course》第 10 章:文件和异常