python让程序更健壮 – 异常处理：最佳实践
欢迎学习异常处理的”艺术”！
目前你已经掌握了异常处理的基本技能，让我们来学习如何优雅地使用它们。好的异常处理就像好的礼仪，让代码更加优雅和专业！
1. 具体异常优于通用异常
不好的做法：
try:
# 各种操作
file = open("data.txt", "r")
number = int("123")
result = 10 / 2
except Exception: # 太宽泛！
print("发生错误")好的做法：
try:
file = open("data.txt", "r")
content = file.read()
number = int(content)
result = 10 / number
except FileNotFoundError:
print("❌ 文件不存在")
except ValueError:
print("❌ 文件内容不是有效数字")
except ZeroDivisionError:
print("❌ 不能除以零")2. 不要静默吞掉异常
不好的做法：
def get_config_value(key):
try:
return config[key]
except KeyError:
pass # 静默失败！没人知道出错了好的做法：
def get_config_value(key):
try:
return config[key]
except KeyError:
logging.warning(f"配置项 '{key}' 不存在，使用默认值")
return DEFAULT_VALUES.get(key)
# 或者重新抛出
# raise ConfigurationError(f"缺少必要的配置项: {key}")3. 使用上下文管理器自动清理资源
不好的做法：
try:
file = open("data.txt", "r")
# 处理文件
content = file.read()
file.close() # 可能忘记调用
except IOError:
if file:
file.close()好的做法：
try:
with open("data.txt", "r") as file:
content = file.read()
# 文件会自动关闭，即使发生异常
except FileNotFoundError:
print("文件不存在")4. 提供有意义的错误信息
不好的做法：
def calculate_bmi(weight, height):
if height <= 0:
raise ValueError("高度无效")好的做法：
def calculate_bmi(weight, height):
if height <= 0:
raise ValueError(
f"高度必须为正数，收到: {height}。"
f"请检查单位是否正确（米为单位）。"
)
if weight <= 0:
raise ValueError(
f"体重必须为正数，收到: {weight}。"
f"请检查单位是否正确（千克为单位）。"
)
return weight / (height ** 2)5. 实际案例：Web API 错误处理
示例1：RESTful API 错误处理
import json
from http import HTTPStatus
class APIError(Exception):
"""API异常基类"""
def __init__(self, message, status_code=HTTPStatus.BAD_REQUEST, error_code=None):
super().__init__(message)
self.message = message
self.status_code = status_code
self.error_code = error_code
def to_dict(self):
return {
"error": {
"code": self.error_code or self.status_code,
"message": self.message
}
}
class ValidationError(APIError):
"""数据验证错误"""
def __init__(self, message, field=None):
super().__init__(message, HTTPStatus.BAD_REQUEST, "VALIDATION_ERROR")
self.field = field
class AuthenticationError(APIError):
"""认证错误"""
def __init__(self, message="认证失败"):
super().__init__(message, HTTPStatus.UNAUTHORIZED, "AUTH_ERROR")
class ResourceNotFoundError(APIError):
"""资源未找到"""
def __init__(self, resource_type, resource_id):
message = f"{resource_type} '{resource_id}' 未找到"
super().__init__(message, HTTPStatus.NOT_FOUND, "NOT_FOUND")
def handle_api_request(request_handler):
"""API请求处理装饰器"""
def wrapper(*args, **kwargs):
try:
return request_handler(*args, **kwargs)
except APIError as e:
# 已知的业务异常
return json.dumps(e.to_dict()), e.status_code
except Exception as e:
# 未知异常，记录日志但不暴露细节给用户
logging.error(f"未处理的异常: {e}", exc_info=True)
error = APIError(
"服务器内部错误",
HTTPStatus.INTERNAL_SERVER_ERROR,
"INTERNAL_ERROR"
)
return json.dumps(error.to_dict()), error.status_code
return wrapper
# 使用示例
@handle_api_request
def get_user_profile(user_id):
if not user_id or not isinstance(user_id, int):
raise ValidationError("用户ID必须为整数", field="user_id")
user = database.get_user(user_id) # 假设的数据库操作
if not user:
raise ResourceNotFoundError("用户", user_id)
return {"user": user}6. 数据库操作的最佳实践 ️
示例2：数据库事务处理
import sqlite3
from contextlib import contextmanager
class DatabaseError(Exception):
"""数据库异常"""
pass
@contextmanager
def database_transaction(db_path):
"""数据库事务上下文管理器"""
connection = None
try:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
yield connection
connection.commit()
print("✅ 事务提交成功")
except sqlite3.Error as e:
if connection:
connection.rollback()
print(" 事务回滚")
raise DatabaseError(f"数据库操作失败: {e}")
finally:
if connection:
connection.close()
print(" 数据库连接已关闭")
def safe_database_operation():
"""安全的数据库操作示例"""
try:
with database_transaction("app.db") as conn:
# 创建表
conn.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL
)
''')
# 插入数据
conn.execute(
"INSERT INTO users (name, email) VALUES (?, ?)",
("Alice", "alice@example.com")
)
# 故意制造错误（重复email）
conn.execute(
"INSERT INTO users (name, email) VALUES (?, ?)",
("Bob", "alice@example.com") # 重复email，会违反UNIQUE约束
)
except DatabaseError as e:
print(f"❌ 数据库错误: {e}")
except Exception as e:
print(f"❌ 未知错误: {e}")
# 测试
safe_database_operation()7. 文件处理的最佳实践
示例3：健壮的文件处理
import os
from pathlib import Path
class FileProcessor:
"""文件处理器"""
@staticmethod
def safe_read_file(file_path, encoding='utf-8'):
"""安全读取文件"""
path = Path(file_path)
# 前置验证
if not path.exists():
raise FileNotFoundError(f"文件不存在: {file_path}")
if not path.is_file():
raise ValueError(f"路径不是文件: {file_path}")
if path.stat().st_size == 0:
raise ValueError(f"文件为空: {file_path}")
try:
with open(file_path, 'r', encoding=encoding) as file:
return file.read()
except UnicodeDecodeError:
# 尝试其他编码
try:
with open(file_path, 'r', encoding='latin-1') as file:
return file.read()
except Exception as e:
raise ValueError(f"无法解码文件: {e}")
except PermissionError:
raise PermissionError(f"没有读取权限: {file_path}")
@staticmethod
def safe_write_file(file_path, content, backup=True):
"""安全写入文件"""
path = Path(file_path)
# 如果需要备份且文件已存在
if backup and path.exists():
backup_path = path.with_suffix('.bak')
try:
path.rename(backup_path)
print(f" 已创建备份: {backup_path}")
except Exception as e:
raise IOError(f"创建备份失败: {e}")
# 确保目录存在
path.parent.mkdir(parents=True, exist_ok=True)
try:
# 使用临时文件避免写入过程中出错导致原文件损坏
temp_path = path.with_suffix('.tmp')
with open(temp_path, 'w', encoding='utf-8') as file:
file.write(content)
# 原子操作：重命名临时文件为目标文件
temp_path.rename(path)
print(f"✅ 文件写入成功: {file_path}")
except Exception as e:
# 清理临时文件
if temp_path.exists():
temp_path.unlink()
raise IOError(f"文件写入失败: {e}")
# 使用示例
processor = FileProcessor()
try:
content = processor.safe_read_file("important_data.txt")
processed_content = content.upper()
processor.safe_write_file("processed_data.txt", processed_content)
except (FileNotFoundError, ValueError, PermissionError, IOError) as e:
print(f"文件处理失败: {e}")8. 配置管理的错误处理 ⚙️
示例4：配置加载器
import yaml
import os
from typing import Dict, Any
class ConfigError(Exception):
"""配置错误"""
pass
class ConfigLoader:
"""配置加载器"""
REQUIRED_KEYS = ['database', 'api_key', 'timeout']
@classmethod
def load_config(cls, config_path: str) -> Dict[str, Any]:
"""加载并验证配置文件"""
try:
with open(config_path, 'r') as file:
config = yaml.safe_load(file)
except FileNotFoundError:
raise ConfigError(f"配置文件不存在: {config_path}")
except yaml.YAMLError as e:
raise ConfigError(f"配置文件格式错误: {e}")
except Exception as e:
raise ConfigError(f"读取配置文件失败: {e}")
# 验证必需配置项
cls._validate_config(config)
return config
@classmethod
def _validate_config(cls, config: Dict[str, Any]):
"""验证配置完整性"""
if not config:
raise ConfigError("配置文件为空")
missing_keys = []
for key in cls.REQUIRED_KEYS:
if key not in config:
missing_keys.append(key)
if missing_keys:
raise ConfigError(f"缺少必需配置项: {', '.join(missing_keys)}")
# 验证具体值
if config.get('timeout') <= 0:
raise ConfigError("超时时间必须大于0")
if not config.get('api_key') or not isinstance(config['api_key'], str):
raise ConfigError("API密钥必须为非空字符串")
# 使用示例
try:
config = ConfigLoader.load_config("app_config.yaml")
print("✅ 配置加载成功")
except ConfigError as e:
print(f"❌ 配置错误: {e}")
# 使用默认配置或退出程序9. 异常处理检查清单 ✅
在编写异常处理代码时，问自己这些问题：
- 我是否捕获了具体的异常类型？
- 我是否提供了有意义的错误信息？
- 我是否妥善清理了资源？
- 我是否记录了重大的异常？
- 我是否思考了所有可能的错误情况？
- 我的异常处理是否有助于调试？
- 我是否避免了静默失败？
- 我是否使用了适当的异常层次结构？
10. 最终实战项目
综合示例：天气预报应用
import requests
import logging
from datetime import datetime
from typing import Optional, Dict, Any
# 自定义异常
class WeatherAPIError(Exception):
"""天气API异常"""
pass
class InvalidLocationError(WeatherAPIError):
"""无效位置异常"""
pass
class WeatherService:
"""天气服务"""
def __init__(self, api_key: str):
if not api_key:
raise ValueError("API密钥不能为空")
self.api_key = api_key
self.base_url = "https://api.weather.com/v1"
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def get_weather(self, city: str, country: str) -> Dict[str, Any]:
"""获取天气信息"""
# 参数验证
if not city or not isinstance(city, str):
raise InvalidLocationError("城市名称必须为非空字符串")
if not country or not isinstance(country, str):
raise InvalidLocationError("国家名称必须为非空字符串")
try:
logging.info(f"获取天气信息: {city}, {country}")
response = requests.get(
f"{self.base_url}/weather",
params={
"city": city,
"country": country,
"apikey": self.api_key
},
timeout=10 # 设置超时
)
# 检查HTTP状态码
response.raise_for_status()
data = response.json()
# 检查API响应状态
if data.get('status') != 'success':
raise WeatherAPIError(f"API返回错误: {data.get('message', '未知错误')}")
return data
except requests.exceptions.Timeout:
raise WeatherAPIError("请求超时，请稍后重试")
except requests.exceptions.ConnectionError:
raise WeatherAPIError("网络连接错误，请检查网络")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
raise InvalidLocationError(f"找不到位置: {city}, {country}")
elif e.response.status_code == 401:
raise WeatherAPIError("API密钥无效")
else:
raise WeatherAPIError(f"HTTP错误: {e}")
except ValueError as e:
raise WeatherAPIError(f"解析响应数据失败: {e}")
except Exception as e:
logging.error(f"获取天气信息时发生未知错误: {e}")
raise WeatherAPIError("服务暂时不可用，请稍后重试")
# 使用示例
def main():
try:
weather_service = WeatherService("your_api_key_here")
weather_data = weather_service.get_weather("Beijing", "China")
print(f"✅ 天气信息: {weather_data}")
except InvalidLocationError as e:
print(f"❌ 位置错误: {e}")
except WeatherAPIError as e:
print(f"❌ 天气服务错误: {e}")
except Exception as e:
print(f"❌ 未知错误: {e}")
if __name__ == "__main__":
main()总结
异常处理的最佳实践：
- 具体明确：捕获具体异常，提供清晰错误信息
- 不要静默：避免无声失败，适当记录和报告
- 资源管理：使用上下文管理器自动清理资源
- 分层处理：在不同层级适当处理异常
- 用户友善：给用户有意义的反馈，给开发者详细的日志
- 防御性编程：预见可能的问题并提前处理
记住： 优秀的异常处理让程序从”能用”变为”好用”，从”脆弱”变为”健壮”！
祝贺你完成了Python异常处理的全部学习！目前你已经具备了编写健壮、专业Python程序的能力！
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
暂无评论...
