当 Python 代码遇到错误时,捕获正确的异常可能意味着正常恢复和突然崩溃之间的区别。让我们探索如何有效地处理多个异常,并提供您可以在自己的项目中使用的实际示例。
了解异常层次结构
在我们深入研究捕获多个异常之前,了解 Python 如何组织其异常会很有帮助。所有内置异常都继承自 'BaseException' 类,最常见的异常继承自 'Exception'。
try:
number = int("abc")
result = 10 / 0
except ValueError as ve:
print(f"ValueError caught: {ve}")
except ZeroDivisionError as zde:
print(f"ZeroDivisionError caught: {zde}")
在此示例中,仅触发 'ValueError',因为 Python 在第一个匹配的异常处停止。代码永远不会到达除以零运算。
单行中的多个异常
当不同的异常需要相同的处理时,您可以使用括号对它们进行分组:
def process_data(data):
try:
if isinstance(data, str):
return int(data)
return 10 / data
except (ValueError, ZeroDivisionError) as error:
print(f"Error processing data: {error}")
return None
# Example usage
print(process_data("abc")) # ValueError
print(process_data(0)) # ZeroDivisionError
print(process_data(5)) # Works fine, returns 2.0
当您希望以相同的方式处理多个异常时,此模式特别有用。“as error”子句捕获特定的异常对象,允许您在需要时访问错误详细信息。
例外顺序很重要
Python 按照您指定的顺序检查异常。这在处理异常继承时变得至关重要:
def read_config():
try:
with open('config.json') as f:
data = json.loads(f.read())
return data['settings']['timeout']
except FileNotFoundError:
print("Config file is missing")
return 30 # default timeout
except json.JSONDecodeError:
print("Config file is malformed")
return 30
except KeyError:
print("Required settings are missing")
return 30
此代码在读取配置文件时处理三种特定的故障模式。每个异常都提供有关出错原因的明确反馈。
使用异常组
对于更复杂的方案,您可能希望对不同的异常组进行不同的处理:
def fetch_and_process_data(url):
try:
response = requests.get(url)
response.raise_for_status()
data = response.json()
return process_data(data)
except (requests.ConnectionError, requests.Timeout) as network_error:
# Handle network-related errors
print(f"Network error: {network_error}")
return None
except (requests.HTTPError, requests.RequestException) as http_error:
# Handle HTTP-related errors
print(f"HTTP error: {http_error}")
return None
except json.JSONDecodeError as parse_error:
# Handle JSON parsing errors
print(f"Data parsing error: {parse_error}")
return None
此示例说明如何按类别组织异常,使错误处理更具逻辑性和可维护性。
使用 Finally 处理异常
有时,您需要确保无论是否发生异常,都运行某些清理代码:
def update_database(connection, data):
cursor = None
try:
cursor = connection.cursor()
cursor.execute("UPDATE users SET status = ?", (data,))
connection.commit()
except sqlite3.Error as db_error:
print(f"Database error: {db_error}")
connection.rollback()
finally:
if cursor:
cursor.close()
“finally”块可确保即使发生异常也会进行资源清理,从而防止资源泄漏。
自定义异常类
创建自定义异常有助于使错误处理更加具体和有意义:
class ValidationError(Exception):
pass
class NetworkError(Exception):
pass
class DataProcessingError(Exception):
def __init__(self, message, raw_data=None):
super().__init__(message)
self.raw_data = raw_data
def process_user_data(data):
try:
if not data:
raise ValidationError("Data cannot be empty")
if not isinstance(data, dict):
raise DataProcessingError("Invalid data format", raw_data=data)
# Process the data...
except ValidationError as ve:
print(f"Validation failed: {ve}")
return False
except DataProcessingError as dpe:
print(f"Processing error: {dpe}")
print(f"Problematic data: {dpe.raw_data}")
return False
自定义异常使您的代码更具表现力,并有助于捕获内置异常未涵盖的特定错误条件。
实际示例:文件处理管道
下面是一个在文件处理管道中组合多种异常处理技术的实际示例:
import json
from pathlib import Path
import csv
from typing import Dict, List
def process_data_files(input_dir: str, output_file: str) -> bool:
try:
# Convert input_dir to Path object
input_path = Path(input_dir)
if not input_path.exists():
raise FileNotFoundError(f"Input directory {input_dir} does not exist")
processed_data = []
# Process each JSON file in the directory
for file_path in input_path.glob("*.json"):
try:
with open(file_path) as f:
data = json.load(f)
processed_data.extend(transform_data(data))
except json.JSONDecodeError:
print(f"Skipping malformed file: {file_path}")
continue
except Exception as e:
print(f"Error processing {file_path}: {e}")
continue
# Write processed data to CSV
write_output(processed_data, output_file)
return True
except (PermissionError, OSError) as e:
print(f"File system error: {e}")
return False
except Exception as e:
print(f"Unexpected error: {e}")
return False
def transform_data(data: List[Dict]) -> List[Dict]:
transformed = []
for item in data:
try:
transformed.append({
'id': item['id'],
'name': item['name'],
'value': float(item['value'])
})
except KeyError as ke:
print(f"Missing required field: {ke}")
except ValueError:
print(f"Invalid value for item {item.get('id', 'unknown')}")
return transformed
def write_output(data: List[Dict], output_file: str) -> None:
if not data:
raise ValueError("No data to write")
with open(output_file, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['id', 'name', 'value'])
writer.writeheader()
writer.writerows(data)
此示例显示了一个完整的数据处理管道,该管道:
- 将文件系统错误与数据处理错误分开处理
- 将嵌套的 try-except 块用于不同的处理阶段
- 即使单个文件失败,也可以继续处理
- 为每种类型的潜在故障实施特定的错误处理
结论
有效的异常处理是指在捕获特定异常和保持代码的可读性和维护性之间找到适当的平衡。请记住以下关键点:
- 当您可以以不同的方式处理特定异常时捕获它们
- 处理相同时对相关异常进行分组
- 将异常处理程序从最具体到最不具体排序
- 使用自定义异常使您的错误处理更具表现力
- 始终清理 'finally' 块中的资源
通过深思熟虑地应用这些模式,您可以编写代码,以优雅地处理错误,同时保持清晰和可维护性。