1. 异常基础
1-1. 异常的概念
异常（Exception）是程序执行过程中出现的错误或意外情况。Python 使用异常处理机制来处理这些情况，而不是传统的错误码方式。
异常处理的主要优势：
- 错误隔离：将错误处理代码与正常业务逻辑分离
- 错误传播：异常可以沿着调用栈向上传播
- 错误恢复：提供错误恢复的机会
- 错误信息：提供详细的错误信息
1-2. 异常层次结构
Python 的异常体系是一个层次结构，所有异常都继承自 BaseException：
BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
├── StopIteration
├── ArithmeticError
│ ├── FloatingPointError
│ ├── OverflowError
│ └── ZeroDivisionError
├── AssertionError
├── AttributeError
├── BufferError
├── EOFError
├── ImportError
├── LookupError
│ ├── IndexError
│ └── KeyError
├── MemoryError
├── NameError
├── OSError
│ ├── FileNotFoundError
│ ├── PermissionError
│ └── TimeoutError
├── RuntimeError
├── SyntaxError
├── TypeError
└── ValueError1-3. 基本异常处理
python
try:
# 可能引发异常的代码
result = 10 / 0
except ZeroDivisionError:
# 处理特定异常
print("除数不能为零")
except Exception as e:
# 处理其他异常
print(f"发生错误: {e}")
else:
# 没有异常时执行
print("计算成功")
finally:
# 无论是否发生异常都会执行
print("清理工作")2. 异常处理进阶
2-1. 自定义异常
python
class ValidationError(Exception):
"""数据验证错误"""
def __init__(self, message, field=None):
self.message = message
self.field = field
super().__init__(self.message)
def __str__(self):
if self.field:
return f"{self.field}: {self.message}"
return self.message
# 使用自定义异常
def validate_age(age):
if age < 0:
raise ValidationError("年龄不能为负数", "age")
if age > 150:
raise ValidationError("年龄不能超过150岁", "age")
return age
try:
validate_age(-1)
except ValidationError as e:
print(e) # 输出: age: 年龄不能为负数2-2. 异常链
python
def process_data(data):
try:
return int(data)
except ValueError as e:
raise ValueError("数据处理失败") from e
try:
process_data("abc")
except ValueError as e:
print(f"错误: {e}")
print(f"原因: {e.__cause__}")2-3. 上下文管理器
python
class DatabaseConnection:
def __init__(self, db_name):
self.db_name = db_name
def __enter__(self):
print(f"连接数据库: {self.db_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type:
print(f"发生错误: {exc_val}")
print(f"关闭数据库连接: {self.db_name}")
return True # 抑制异常
# 使用上下文管理器
with DatabaseConnection("mydb") as db:
print("执行数据库操作")
raise ValueError("测试错误")3. 异常处理最佳实践
3-1. 异常处理原则
- 具体异常优先
python
try:
file = open("data.txt")
except FileNotFoundError:
print("文件不存在")
except PermissionError:
print("没有权限")
except Exception:
print("其他错误")- 避免空的 except 块
python
# 不好的做法
try:
process_data()
except:
pass
# 好的做法
try:
process_data()
except Exception as e:
logging.error(f"处理数据失败: {e}")- 使用 finally 清理资源
python
file = None
try:
file = open("data.txt")
process_file(file)
except IOError as e:
print(f"文件操作失败: {e}")
finally:
if file:
file.close()3-2. 异常处理模式
- 重试机制
python
import time
from typing import Callable, Type, Tuple
def retry(
func: Callable,
exceptions: Tuple[Type[Exception], ...],
max_attempts: int = 3,
delay: float = 1.0
) -> Callable:
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
print(f"尝试 {attempt + 1} 失败: {e}")
time.sleep(delay)
return wrapper
@retry(exceptions=(ConnectionError, TimeoutError))
def fetch_data():
# 模拟网络请求
raise ConnectionError("连接失败")- 异常转换
python
class DataProcessor:
def process(self, data):
try:
return self._process_data(data)
except ValueError as e:
raise ProcessingError("数据处理失败") from e
except TypeError as e:
raise ProcessingError("数据类型错误") from e
def _process_data(self, data):
# 实际的数据处理逻辑
pass- 异常日志
python
import logging
import traceback
logging.basicConfig(level=logging.ERROR)
def process_with_logging(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(f"错误: {e}")
logging.error(traceback.format_exc())
raise
return wrapper4. 常见异常处理场景
4-1. 文件操作异常
python
def read_file(filename):
try:
with open(filename, 'r') as file:
return file.read()
except FileNotFoundError:
print(f"文件 {filename} 不存在")
except PermissionError:
print(f"没有权限读取文件 {filename}")
except IOError as e:
print(f"读取文件时发生错误: {e}")4-2. 网络请求异常
python
import requests
from requests.exceptions import RequestException
def fetch_url(url):
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.text
except requests.Timeout:
print("请求超时")
except requests.ConnectionError:
print("连接错误")
except requests.HTTPError as e:
print(f"HTTP错误: {e}")
except RequestException as e:
print(f"请求失败: {e}")4-3. 数据库操作异常
python
import sqlite3
def query_database(query):
conn = None
try:
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
cursor.execute(query)
return cursor.fetchall()
except sqlite3.OperationalError as e:
print(f"数据库操作错误: {e}")
except sqlite3.IntegrityError as e:
print(f"数据完整性错误: {e}")
finally:
if conn:
conn.close()5. 调试技巧
5-1. 异常信息获取
python
import sys
import traceback
try:
# 可能出错的代码
result = 1 / 0
except Exception as e:
# 获取异常类型
exc_type = type(e).__name__
# 获取异常信息
exc_message = str(e)
# 获取异常堆栈
exc_traceback = traceback.format_exc()
# 获取异常发生的文件位置
exc_file = sys.exc_info()[2].tb_frame.f_code.co_filename
exc_line = sys.exc_info()[2].tb_lineno
print(f"异常类型: {exc_type}")
print(f"异常信息: {exc_message}")
print(f"异常位置: {exc_file}:{exc_line}")
print(f"堆栈跟踪:\n{exc_traceback}")5-2. 断言调试
python
def calculate_discount(price, discount):
assert price > 0, "价格必须大于0"
assert 0 <= discount <= 1, "折扣必须在0到1之间"
return price * (1 - discount)
# 使用断言
try:
result = calculate_discount(-100, 0.2)
except AssertionError as e:
print(f"断言失败: {e}")6. 性能考虑
6-1. 异常处理开销
python
import timeit
def without_exception():
result = []
for i in range(1000):
if i % 2 == 0:
result.append(i)
return result
def with_exception():
result = []
for i in range(1000):
try:
if i % 2 == 0:
result.append(i)
except:
pass
return result
# 测试性能
print("无异常处理:", timeit.timeit(without_exception, number=1000))
print("有异常处理:", timeit.timeit(with_exception, number=1000))6-2. 异常处理优化
python
# 不好的做法
def process_data(data):
try:
return int(data)
except:
return None
# 好的做法
def process_data(data):
if isinstance(data, (int, float)):
return int(data)
if isinstance(data, str) and data.isdigit():
return int(data)
return None7. 实战示例
7-1. 数据验证器
python
from typing import Any, Dict, List
from dataclasses import dataclass
@dataclass
class ValidationResult:
is_valid: bool
errors: List[str]
class DataValidator:
def __init__(self):
self.rules = {}
def add_rule(self, field: str, rule: callable, message: str):
if field not in self.rules:
self.rules[field] = []
self.rules[field].append((rule, message))
def validate(self, data: Dict[str, Any]) -> ValidationResult:
errors = []
for field, rules in self.rules.items():
value = data.get(field)
for rule, message in rules:
try:
if not rule(value):
errors.append(f"{field}: {message}")
except Exception as e:
errors.append(f"{field}: 验证失败 - {str(e)}")
return ValidationResult(len(errors) == 0, errors)
# 使用示例
validator = DataValidator()
validator.add_rule("age", lambda x: x >= 0, "年龄不能为负数")
validator.add_rule("email", lambda x: "@" in x, "邮箱格式不正确")
data = {"age": -1, "email": "invalid"}
result = validator.validate(data)
if not result.is_valid:
print("验证失败:")
for error in result.errors:
print(f"- {error}")7-2. 重试装饰器
python
from functools import wraps
import time
import random
from typing import Type, Tuple, Callable, Any
class RetryError(Exception):
"""重试失败异常"""
pass
def retry(
exceptions: Tuple[Type[Exception], ...] = (Exception,),
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
jitter: bool = True
) -> Callable:
"""
重试装饰器
Args:
exceptions: 需要重试的异常类型
max_attempts: 最大重试次数
delay: 初始延迟时间（秒）
backoff: 延迟时间增长因子
jitter: 是否添加随机抖动
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
current_delay = delay
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:
raise RetryError(f"重试 {max_attempts} 次后仍然失败") from e
# 计算下一次延迟时间
if jitter:
current_delay = current_delay * backoff * (0.5 + random.random())
else:
current_delay = current_delay * backoff
print(f"尝试 {attempt + 1} 失败: {e}")
print(f"等待 {current_delay:.2f} 秒后重试...")
time.sleep(current_delay)
raise RetryError("重试失败") from last_exception
return wrapper
return decorator
# 使用示例
@retry(exceptions=(ConnectionError, TimeoutError), max_attempts=3)
def unreliable_function():
if random.random() < 0.7:
raise ConnectionError("连接失败")
return "成功"
try:
result = unreliable_function()
print(f"结果: {result}")
except RetryError as e:
print(f"最终失败: {e}")TIP
异常处理最佳实践：
- 使用具体的异常类型而不是捕获所有异常
- 提供有意义的错误信息
- 适当使用异常链
- 清理资源使用 finally 或上下文管理器
- 考虑性能影响
- 记录异常信息
- 实现适当的重试机制