📢 转载信息
原文链接：https://www.kdnuggets.com/5-useful-diy-python-functions-for-error-handling
原文作者：Bala Priya C
Image by Author
# 简介
在看似稳固的代码中，错误处理常常是薄弱环节。在实际项目中，诸如缺少键、请求失败和运行时间过长的函数等问题会频繁出现。Python 内置的 try-except 块虽然有用，但它们本身并不能涵盖许多实际场景。
您需要将常见的失败场景封装到小型、可重用的函数中，这些函数有助于处理带有次数限制的重试、输入验证以及防止代码运行时间过长的保护措施。本文将介绍五个可用于网页抓取、构建应用程序接口 (API)、处理用户数据等任务的错误处理函数。
# 使用指数退避重试失败的操作
在许多项目中，API 调用和网络请求经常会失败。初学者的做法是尝试一次，捕获任何异常，记录下来，然后停止。更好的方法是重试。
这时就该指数退避 (exponential backoff) 发挥作用了。不要立即用密集的重试来轰炸一个失败的服务（这只会让情况更糟），而是在每次尝试之间等待更长的时间：1 秒，然后 2 秒，然后 4 秒，以此类推。
让我们来构建一个实现此功能的装饰器：
import time import functools from typing import Callable, Type, Tuple def retry_with_backoff( max_attempts: int = 3, base_delay: float = 1.0, exponential_base: float = 2.0, exceptions: Tuple[Type[Exception], ...] = (Exception,) ): """ Retry a function with exponential backoff. Args: max_attempts: Maximum number of retry attempts base_delay: Initial delay in seconds exponential_base: Multiplier for delay (2.0 = double each time) exceptions: Tuple of exception types to catch and retry """ def decorator(func: Callable): @functools.wraps(func) def wrapper(*args, **kwargs): 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: delay = base_delay * (exponential_base ** attempt) print(f"Attempt {attempt + 1} failed: {e}") print(f"Retrying in {delay:.1f} seconds...") time.sleep(delay) else: print(f"All {max_attempts} attempts failed") raise last_exception return wrapper return decorator
该装饰器包装了您的函数，并捕获指定的异常。核心计算是 delay = base_delay * (exponential_base ** attempt)。当 base_delay=1 和 exponential_base=2 时，您的延迟为 1 秒、2 秒、4 秒、8 秒。这为压力过大的系统提供了恢复时间。
exceptions 参数允许您指定要重试哪些错误。您可能会重试 ConnectionError，但不会重试 ValueError，因为连接问题是暂时的，而验证错误则不是。
现在让我们看看它的实际应用：
import random @retry_with_backoff(max_attempts=4, base_delay=0.5, exceptions=(ConnectionError,)) def fetch_user_data(user_id): """Simulate an unreliable API.""" if random.random() < 0.6: # 60% failure rate raise ConnectionError("Service temporarily unavailable") return {"id": user_id, "name": "Sara", "status": "active"} # Watch it retry automatically result = fetch_user_data(12345) print(f"Success: {result}")
输出：
Success: {'id': 12345, 'name': 'Sara', 'status': 'active'}
# 使用可组合规则验证输入
用户输入验证非常繁琐且重复。您需要检查字符串是否为空、数字是否在范围内、电子邮件是否有效。不知不觉中，您的代码中到处都是嵌套的 if 语句，看起来一团糟。
让我们构建一个易于使用的验证系统。首先，我们需要一个自定义异常：
from typing import Any, Callable, Dict, List, Optional class ValidationError(Exception): """Raised when validation fails.""" def __init__(self, field: str, errors: List[str]): self.field = field self.errors = errors super().__init__(f"{field}: {', '.join(errors)}")
此异常包含多个错误消息。当验证失败时，我们希望向用户显示所有错误之处，而不仅仅是第一个错误。
现在来看看验证器：
def validate_input( value: Any, field_name: str, rules: Dict[str, Callable[[Any], bool]], messages: Optional[Dict[str, str]] = None ) -> Any: """ Validate input against multiple rules. Returns the value if valid, raises ValidationError otherwise. """ if messages is None: messages = {} errors = [] for rule_name, rule_func in rules.items(): try: if not rule_func(value): error_msg = messages.get( rule_name, f"Failed validation rule: {rule_name}" ) errors.append(error_msg) except Exception as e: errors.append(f"Validation error in {rule_name}: {str(e)}") if errors: raise ValidationError(field_name, errors) return value
在 rules 字典中，每个规则只是一个返回 True 或 False 的函数。这使得规则具有可组合性和可重用性。
让我们创建一些常见的验证规则：
# Reusable validation rules def not_empty(value: str) -> bool: return bool(value and value.strip()) def min_length(min_len: int) -> Callable: return lambda value: len(str(value)) >= min_len def max_length(max_len: int) -> Callable: return lambda value: len(str(value)) <= max_len def in_range(min_val: float, max_val: float) -> Callable: return lambda value: min_val <= float(value) <= max_val
请注意，min_length、max_length 和 in_range 如何成为工厂函数。它们返回带有特定参数配置的验证函数。这样您就可以编写 min_length(3)，而不是为每个长度要求创建一个新函数。
让我们验证一个用户名：
try: username = validate_input( "ab", "username", { "not_empty": not_empty, "min_length": min_length(3), "max_length": max_length(20), }, messages={ "not_empty": "Username cannot be empty", "min_length": "Username must be at least 3 characters", "max_length": "Username cannot exceed 20 characters", } ) print(f"Valid username: {username}") except ValidationError as e: print(f"Invalid: {e}")
输出：
Invalid: username: Username must be at least 3 characters
这种方法具有良好的可扩展性。一次定义您的规则，以任何您需要的方式组合它们，并获得清晰的错误消息。
# 安全地导航嵌套字典
访问嵌套字典通常很困难。当键不存在时，您会收到 KeyError；当您尝试对字符串进行下标访问时，会收到 TypeError；您的代码会充斥着连续的 .get() 调用或防御性的 try-except 块。处理来自 API 的 JavaScript Object Notation (JSON) 使这更具挑战性。
让我们构建一个可以安全导航嵌套结构的函数：
from typing import Any, Optional, List, Union def safe_get( data: dict, path: Union[str, List[str]], default: Any = None, separator: str = "." ) -> Any: """ Safely get a value from a nested dictionary. Args: data: The dictionary to access path: Dot-separated path (e.g., "user.address.city") or list of keys default: Value to return if path doesn't exist separator: Character to split path string (default: ".") Returns: The value at the path, or default if not found """ # Convert string path to list if isinstance(path, str): keys = path.split(separator) else: keys = path current = data for key in keys: try: # Handle list indices (convert string to int if numeric) if isinstance(current, list): try: key = int(key) except (ValueError, TypeError): return default current = current[key] except (KeyError, IndexError, TypeError): return default return current
该函数将路径拆分成单个键，并逐步导航嵌套结构。如果任何键不存在，或者您尝试访问非下标对象，它将返回默认值而不是崩溃。
它还可以自动处理列表索引。如果当前值是列表，并且键是数字，它会将键转换为整数。
这是用于设置值的伴随函数：
def safe_set( data: dict, path: Union[str, List[str]], value: Any, separator: str = ".", create_missing: bool = True ) -> bool: """ Safely set a value in a nested dictionary. Args: data: The dictionary to modify path: Dot-separated path or list of keys value: Value to set separator: Character to split path string create_missing: Whether to create missing intermediate dicts Returns: True if successful, False otherwise """ if isinstance(path, str): keys = path.split(separator) else: keys = path if not keys: return False current = data # Navigate to the parent of the final key for key in keys[:-1]: if key not in current: if create_missing: current[key] = {} else: return False current = current[key] if not isinstance(current, dict): return False # Set the final value current[keys[-1]] = value return True
safe_set 函数会按需创建嵌套结构并设置值。这对于动态构建字典非常有用。
让我们同时测试一下：
# Sample nested data user_data = { "user": { "name": "Anna", "address": { "city": "San Francisco", "zip": "94105" }, "orders": [ {"id": 1, "total": 99.99}, {"id": 2, "total": 149.50} ] } } # Safe get examples city = safe_get(user_data, "user.address.city") print(f"City: {city}") country = safe_get(user_data, "user.address.country", default="Unknown") print(f"Country: {country}") first_order = safe_get(user_data, "user.orders.0.total") print(f"First order: ${first_order}") # Safe set example new_data = {} safe_set(new_data, "user.settings.theme", "dark") print(f"Created: {new_data}")
输出：
City: San Francisco Country: Unknown First order: $99.99 Created: {'user': {'settings': {'theme': 'dark'}}}
这种模式消除了防御性编程的混乱，使您在处理 JSON、配置文件或任何深度嵌套数据时代码更加整洁。
# 对长时间运行的操作强制执行超时
有些操作耗时过长。数据库查询可能会挂起，网页抓取操作可能会卡在慢速服务器上，或者计算可能会无限运行。您需要一种方法来设置时间限制并退出。
这是一个使用线程的超时装饰器：
import threading import functools from typing import Callable, Optional class TimeoutError(Exception): """Raised when an operation exceeds its timeout.""" pass def timeout(seconds: int, error_message: Optional[str] = None): """ Decorator to enforce a timeout on function execution. Args: seconds: Maximum execution time in seconds error_message: Custom error message for timeout """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): result = [TimeoutError( error_message or f"Operation timed out after {seconds} seconds" )] def target(): try: result[0] = func(*args, **kwargs) except Exception as e: result[0] = e thread = threading.Thread(target=target) thread.daemon = True thread.start() thread.join(timeout=seconds) if thread.is_alive(): raise TimeoutError( error_message or f"Operation timed out after {seconds} seconds" ) if isinstance(result[0], Exception): raise result[0] return result[0] return wrapper return decorator
此装饰器在一个单独的线程中运行您的函数，并使用 thread.join(timeout=seconds) 进行等待。如果在超时后线程仍然存活，我们就知道它花费的时间太长，并引发 TimeoutError。
函数结果存储在一个列表中（可变容器）中，以便内部线程可以修改它。如果在线程中发生异常，我们将在主线程中重新引发它。
⚠️ 一个限制：即使在超时后，线程仍在后台运行。对于大多数用例来说这还可以，但对于具有副作用的操作，请小心。
让我们测试一下：
import time @timeout(2, error_message="Query took too long") def slow_database_query(): """Simulate a slow query.""" time.sleep(5) return "Query result" @timeout(3) def fetch_data(): """Simulate a quick operation.""" time.sleep(1) return {"data": "value"} # Test timeout try: result = slow_database_query() print(f"Result: {result}") except TimeoutError as e: print(f"Timeout: {e}") # Test success try: data = fetch_data() print(f"Success: {data}") except TimeoutError as e: print(f"Timeout: {e}")
输出：
Timeout: Query took too long Success: {'data': 'value'}
这种模式对于构建响应式应用程序至关重要。当您抓取网站、调用外部 API 或运行用户代码时，超时可以防止您的程序无限期地挂起。
# 使用自动清理管理资源
打开文件、数据库连接和网络套接字需要仔细清理。如果发生异常，您需要确保资源得到释放。使用 with 语句的上下文管理器可以处理这个问题，但有时您需要更多的控制。
让我们构建一个用于自动资源清理的灵活上下文管理器：
from contextlib import contextmanager from typing import Callable, Any, Optional import traceback @contextmanager def managed_resource( acquire: Callable[[], Any], release: Callable[[Any], None], on_error: Optional[Callable[[Exception, Any], None]] = None, suppress_errors: bool = False ): """ Context manager for automatic resource acquisition and cleanup. Args: acquire: Function to acquire the resource release: Function to release the resource on_error: Optional error handler suppress_errors: Whether to suppress exceptions after cleanup """ resource = None try: resource = acquire() yield resource except Exception as e: if on_error and resource is not None: try: on_error(e, resource) except Exception as handler_error: print(f"Error in error handler: {handler_error}") if not suppress_errors: raise finally: if resource is not None: try: release(resource) except Exception as cleanup_error: print(f"Error during cleanup: {cleanup_error}") traceback.print_exc()
managed_resource 函数是一个上下文管理器工厂。它接受两个必需的函数：一个用于获取资源，一个用于释放资源。release 函数总是在 finally 块中运行，保证即使发生异常也能进行清理。
可选的 on_error 参数允许您在错误传播之前进行处理。这对于日志记录、发送警报或尝试恢复非常有用。suppress_errors 标志决定了异常是被显式引发还是被抑制。
这是一个演示资源跟踪的辅助类：
class ResourceTracker: """Helper class to track resource operations.""" def __init__(self, name: str, verbose: bool = True): self.name = name self.verbose = verbose self.operations = [] def log(self, operation: str): self.operations.append(operation) if self.verbose: print(f"[{self.name}] {operation}") def acquire(self): self.log("Acquiring resource") return self def release(self): self.log("Releasing resource") def use(self, action: str): self.log(f"Using resource: {action}")
让我们测试一下上下文管理器：
# Example: Operation with error handling tracker = ResourceTracker("Database") def error_handler(exception, resource): resource.log(f"Error occurred: {exception}") resource.log("Attempting rollback") try: with managed_resource( acquire=lambda: tracker.acquire(), release=lambda r: r.release(), on_error=error_handler ) as db: db.use("INSERT INTO users") raise ValueError("Duplicate entry") except ValueError as e: print(f"Caught: {e}")
输出：
[Database] Acquiring resource [Database] Using resource: INSERT INTO users [Database] Error occurred: Duplicate entry [Database] Attempting rollback [Database] Releasing resource Caught: Duplicate entry
这种模式对于管理数据库连接、文件句柄、网络套接字、锁以及任何需要保证清理的资源都很有用。它可以防止资源泄漏，并使您的代码更安全。
# 总结
本文中的每个函数都解决了特定的错误处理挑战：重试瞬态故障、系统地验证输入、安全地访问嵌套数据、防止挂起的操作以及管理资源清理。
这些模式在 API 集成、数据处理管道、网页抓取和面向用户的应用程序中反复出现。
这里介绍的技巧使用装饰器、上下文管理器和可组合函数，使错误处理不那么重复且更可靠。您可以直接将这些函数放入您的项目中，或根据您的具体需求进行调整。它们是独立的，易于理解，并且可以解决您经常会遇到的问题。编码愉快！
Bala Priya C 来自印度，是一名开发者和技术作家。她喜欢在数学、编程、数据科学和内容创作的交叉领域工作。她的兴趣和专长领域包括 DevOps、数据科学和自然语言处理。她喜欢阅读、写作、编码和喝咖啡！目前，她正在通过撰写教程、操作指南、观点文章等来学习并与开发者社区分享她的知识。Bala 还创建了引人入胜的资源概述和编码教程。
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区