摘要:def divide_numbers(a, b): try: result = a / b # Convert result to integer return int(result) except zeroDivisionError: print("Erro
当您编写 Python 代码时,事情可能会以许多不同的方式出错。让我们看看如何有效地捕获和处理多个异常,并介绍您在日常编程中会遇到的真实示例。
在 Python 中,您可以捕获不同类型的异常并适当地处理每个异常。下面是一个基本示例:
def divide_numbers(a, b): try: result = a / b # Convert result to integer return int(result) except zeroDivisionError: print("Error: Division by zero!") return None except ValueError: print("Error: Couldn't convert to integer!") return None# Example usage:print(divide_numbers(10, 2)) # Output: 5print(divide_numbers(10, 0)) # Output: Error: Division by zero! Noneprint(divide_numbers(10.5, 2)) # Output: 5让我们来分析一下:
- “try”块包含可能引发异常的代码
- 每个 'except' 块处理特定类型的异常
- 发生异常时,Python 会查找匹配的 'except' 块
- 如果未找到匹配的块,则异常未处理并向上传播
有时,应以相同的方式处理不同的异常。Python 允许您对它们进行分组:
def process_data(data): try: # Try to convert data to float and process it value = float(data) result = 100 / value return result except (ValueError, ZeroDivisionError) as e: # Handle both conversion errors and division by zero print(f"Error processing data: {str(e)}") return None except TypeError as e: print(f"Invalid data type: {str(e)}") return None# Example usage:print(process_data("10")) # Output: 10.0print(process_data("zero")) # Output: Error processing data: could not convert...print(process_data("0")) # Output: Error processing data: division by zeroprint(process_data(None)) # Output: Invalid data type: float argument...“as e”语法捕获异常对象,使您能够访问错误消息和其他详细信息。
Python 的异常形成一个层次结构。您可以捕获特定的异常或其父类:
def read_configuration(Filename): try: with open(filename) as f: data = f.read config = eval(data) # Don't do this in real code! Used for example only return config except FileNotFoundError: # Specific: Handle missing file print(f"Config file '{filename}' not found") return {} except OSError as e: # Parent: Handle other OS-related errors print(f"OS error occurred: {e}") return {} except Exception as e: # Catch-all: Handle unexpected errors print(f"Unexpected error: {e}") return {}# Example usage:config = read_configuration("nonexistent.conf") # Output: Config file 'nonexistent.conf' not found顺序很重要!首先放置更具体的例外,然后是更通用的例外。如果你把 'Exception' 放在前面,它会捕获所有内容,而其他处理程序永远不会运行。
'else' 和 'finally' 块为异常处理添加了更多控制:
def update_user_preferences(user_id, preferences): db_connection = None try: db_connection = connect_to_database # Hypothetical function current_prefs = get_preferences(user_id) current_prefs.update(preferences) save_preferences(user_id, current_prefs) except ConnectionError: print("Database connection failed") return False except KeyError: print("Invalid user ID") return False else: # Runs only if no exception occurred print("Preferences updated successfully") return True finally: # Always runs, whether exception occurred or not if db_connection: db_connection.close# Example usage:success = update_user_preferences(123, {"theme": "dark"})'else' 块仅在未发生异常时运行,因此非常适合仅在成功完成 'try' 块后才应执行的代码。'finally' 块始终运行,使其成为清理代码的理想选择。
对于较大的应用程序,自定义异常有助于创建清晰的错误处理结构:
class DataProcessingError(Exception): """Base class for data processing exceptions""" passclass DataFormatError(DataProcessingError): """Raised when data format is invalid""" passclass DataValidationError(DataProcessingError): """Raised when data validation fails""" passdef process_user_data(data): try: if not isinstance(data, dict): raise DataFormatError("Data must be a dictionary") if "age" not in data: raise DataValidationError("Age is required") if not isinstance(data["age"], int): raise DataFormatError("Age must be an integer") if data["age"]自定义异常使您的错误处理更加具体和有意义。它们还可以帮助其他开发人员了解在使用您的代码时可能会出现什么问题。
下面是一个结合了许多异常处理概念的实际示例:
import jsonimport csvfrom pathlib import Pathdef convert_json_to_csv(input_path, output_path): try: # Check if input file exists if not Path(input_path).exists: raise FileNotFoundError(f"Input file not found: {input_path}") # Read JSON data with open(input_path, 'r') as json_file: try: data = json.load(JSON_file) except json.JSONDecodeError as e: raise DataFormatError(f"Invalid JSON format: {str(e)}") # Ensure data is a list of dictionaries if not isinstance(data, list): raise DataFormatError("JSON data must be a list of records") if not data: raise DataValidationError("JSON data is empty") # Get field names from first record fieldnames = data[0].keys # Write CSV file with open(output_path, 'w', newline='') as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader writer.writerows(data) except (FileNotFoundError, DataFormatError, DataValidationError) as e: print(f"Error: {str(e)}") return False except Exception as e: print(f"Unexpected error: {str(e)}") return False else: print(f"Successfully converted {input_path} to {output_path}") return True# Example usage:input_file = "data.json"output_file = "output.csv"success = convert_json_to_csv(input_file, output_file)此示例说明如何:
- 处理多个特定异常
- 使用自定义异常类
- 安全地执行文件操作
- 提供有意义的错误消息
- 使用 'else' 块进行成功报告
请注意异常处理如何通过以下方式使代码更加健壮和用户友好:
1. 检查输入文件是否存在
2. 验证 JSON 格式
3. 确保数据满足要求
4. 优雅地处理意外错误
5. 就问题所在提供明确的反馈
上下文管理器('with' 语句)自动处理异常和清理。以下是创建自己的方法:
class DatabaseConnection: def __init__(self, connection_string): self.connection_string = connection_string self.connection = None def __enter__(self): try: # Simulate database connection print(f"Connecting to database: {self.connection_string}") self.connection = True return self except Exception as e: raise ConnectionError(f"Failed to connect: {str(e)}") def __exit__(self, exc_type, exc_val, exc_tb): # This runs even if an exception occurs if self.connection: print("Closing database connection") self.connection = None # Return False to propagate exceptions, True to suppress them return Falsedef process_user_records(users): with DatabaseConnection("postgresql://localhost:5432/users") as db: try: for user in users: print(f"Processing user: {user}") # Simulate some database operations if not isinstance(user, dict): raise TypeError("User must be a dictionary") except TypeError as e: print(f"Invalid user data: {e}") return False return True# Example usage:users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}, "invalid"]process_user_records(users)上下文管理器确保资源得到正确清理,即使发生异常也是如此。这特别适用于:
- 数据库连接
- 文件句柄
- 网络连接
- 系统资源
有时你想捕获一个异常并引发一个不同的异常,同时保留原始错误:
class ConfigurationError(Exception): passdef load_config(filename): try: with open(filename) as f: config_text = f.read try: return json.loads(config_text) except json.JSONDecodeError as e: # Raise new exception but keep original error info raise ConfigurationError("Invalid configuration format") from e except FileNotFoundError as e: raise ConfigurationError("Configuration file missing") from edef initialize_application: try: config = load_config("config.json") print("Application initialized") except ConfigurationError as e: print(f"Failed to initialize: {e}") # Access original exception if it exists if e.__cause__: print(f"Original error: {e.__cause__}")# Example usage:initialize_application异常链接有助于维护错误跟踪,同时向用户或日志记录系统提供更有意义的错误。
处理多个异步异常使用异步代码时,异常处理变得更加复杂:
import asynciofrom typing import Listasync def fetch_user(user_id: int) -> dict: # Simulate API call await asyncio.sleep(1) if user_id此模式允许您:
- 同时处理多个操作
- 收集成功的结果和错误
- 即使某些操作失败,也可以继续处理
装饰器可以向多个函数添加一致的异常处理:
from functools import wrapsimport logginglogging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)def handle_exceptions(retries=3, allowed_exceptions=(ConnectionError,)): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): attempts = 0 while attempts此装饰器模式:
- 新增特定异常的重试逻辑
- 持续记录错误
- 保留原始函数的签名和文档
- 可在多个功能之间重复使用
使用线程时,异常需要特殊处理:
import threadingfrom queue import Queueimport tracebackclass WorkerThread(threading.Thread): def __init__(self, task_queue: Queue, error_queue: Queue): super.__init__ self.task_queue = task_queue self.error_queue = error_queue def run(self): while True: try: task = self.task_queue.get if task is None: # Poison pill break # Process task result = self.process_task(task) print(f"Processed task: {result}") except Exception as e: # Capture full exception info error_info = { 'error': e, 'traceback': traceback.format_exc, 'task': task } self.error_queue.put(error_info) finally: self.task_queue.task_done def process_task(self, task): if task此线程异常处理模式:
- 捕获工作线程中的异常
- 保留完整的回溯
- 允许主线程监控和处理 worker 错误
- 提供干净的关机机制
- 扩展到多个工作程序
这些高级模式展示了 Python 的异常处理系统如何处理复杂的实际场景,同时保持代码的清晰度和可靠性。
来源:自由坦荡的湖泊AI一点号