错误处理往往是优秀代码中最薄弱的环节。像缺失主键、请求失败、函数运行时间过长等问题在实际项目中频繁出现。Python自带的try-except块虽然有用但单靠它并不能覆盖许多实际场景。你需要将常见的失败情形封装成小型、可复用的函数以支持带限制的重试、输入验证以及防止代码运行超时的保护机制。本文介绍的五个错误处理函数可用于网页抓取、API 构建、用户数据处理等任务。使用指数退避重试失败操作在许多项目中API 调用和网络请求经常失败。初学者的做法是只尝试一次捕获异常并记录然后停止。更好的做法是重试。这时指数退避就派上了用场。与其立即对失效服务发起多次重试这只会让情况更糟不如在每次尝试之间等待更长的时间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,), ): 使用指数退避算法重试函数的装饰器。 参数: max_attempts: 最大重试次数 base_delay: 初始等待时间秒 exponential_base: 指数基数2.0 表示每次等待时间翻倍 exceptions: 需要捕获并重试的异常类型元组 def decorator(func: Callable): functools.wraps(func) def wrapper(*args, **kwargs): last_exception None # 尝试执行函数最多执行 max_attempts 次 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 1} 次尝试失败: {e}) print(f将在 {delay:.1f} 秒后重试...) time.sleep(delay) else: # 所有尝试均失败抛出最后一次异常 print(f所有 {max_attempts} 次尝试均失败) raise last_exception # 理论上不会执行到这里因为要么返回结果要么抛出异常 # 但为了类型检查器的完整性保留此结构 return wrapper return decorator这个装饰器会包装你的函数并捕获指定的异常。关键计算是delay base_delay * (exponential_base ** attempt)。当base_delay1、exponential_base2时延迟时间为 1s、2s、4s、8s。这给了压力下的系统恢复的时间。exceptions参数让你可以指定哪些错误需要重试。例如你可以重试ConnectionError但不重试ValueError因为连接问题是暂时的而验证错误不是。下面来看它的实际使用if __name__ __main__: counter 0 retry_with_backoff(max_attempts3, base_delay0.5, exceptions(ValueError,)) def unstable_function(): global counter counter 1 print(f正在执行函数当前调用次数: {counter}) if counter 3: raise ValueError(模拟临时性错误) return 成功 try: result unstable_function() print(f最终结果: {result}) except Exception as e: print(f未处理的异常: {e})使用可组合规则验证输入用户输入验证既繁琐又重复。你需要检查字符串是否为空、数字是否在范围内、邮箱格式是否正确。不知不觉中你的代码里就塞满了嵌套的 if 语句变得一团糟。我们来构建一个简单易用的验证系统。首先需要一个自定义异常from typing import Any, Callable, Dict, List, Optional class ValidationError(Exception): 当验证失败时抛出的异常。 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: 根据多条规则验证输入。 参数: value: 待验证的值 field_name: 字段名称用于错误提示 rules: 字典键为规则名称值为验证函数 (接受值返回布尔值) messages: 可选字典键为规则名称值为自定义错误消息 返回: 如果验证通过返回原始值 异常: ValidationError: 如果任何规则验证失败 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未通过验证规则: {rule_name} ) errors.append(error_msg) except Exception as e: # 如果在执行验证规则过程中发生异常例如类型转换错误也视为验证失败 errors.append(f规则 {rule_name} 执行出错: {str(e)}) # 如果存在错误抛出验证异常 if errors: raise ValidationError(field_name, errors) return value在rules字典中每个规则都是一个返回True或False的函数。这使得规则可组合、可复用。我们来创建一些常用的验证规则# --- 可复用的验证规则工厂函数 --- def not_empty(value: Any) - bool: 检查值是否非空针对字符串去除空白后 if isinstance(value, str): return bool(value.strip()) return bool(value) def min_length(min_len: int) - Callable[[Any], bool]: 生成一个检查最小长度的验证函数 def _validator(value: Any) - bool: try: return len(str(value)) min_len except Exception: return False return _validator def max_length(max_len: int) - Callable[[Any], bool]: 生成一个检查最大长度的验证函数 def _validator(value: Any) - bool: try: return len(str(value)) max_len except Exception: return False return _validator def in_range(min_val: float, max_val: float) - Callable[[Any], bool]: 生成一个检查数值范围的验证函数 def _validator(value: Any) - bool: try: num_value float(value) return min_val num_value max_val except (ValueError, TypeError): # 如果无法转换为浮点数则验证失败 return False return _validator注意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: 用户名不能为空, min_length: 用户名长度至少为 3 个字符, max_length: 用户名长度不能超过 20 个字符, } ) print(f验证通过的用户名: {username}) except ValidationError as e: print(f验证失败: {e})输出验证失败: username: 用户名长度至少为 3 个字符这种方法具有良好的扩展性。只需定义一次规则然后按需组合就能得到清晰的错误信息。安全地访问嵌套字典访问嵌套字典常常令人头疼。当键不存在时会抛出KeyError试图对字符串进行下标操作会抛出TypeError而且代码中会充斥着一连串的.get()调用或防御性的 try-except 块。处理来自 API 的 JSON 数据时这个问题尤为突出。我们来构建一个能安全访问嵌套结构的函数from typing import Any, Optional, List, Union def safe_get( data: dict, path: Union[str, List[str]], default: Any None, separator: str . ) - Any: 从嵌套字典中安全地获取值。 参数: data: 要访问的字典 path: 点分路径 (例如 user.address.city) 或键名列表 default: 如果路径不存在时返回的默认值 separator: 用于分割路径字符串的分隔符 (默认: .) 返回: 路径对应的值如果未找到则返回 default # 如果路径是字符串将其分割为键名列表 if isinstance(path, str): keys path.split(separator) else: keys path current data for key in keys: try: # 处理列表索引如果当前层级是列表尝试将键转换为整数索引 if isinstance(current, list): try: index int(key) current current[index] except (ValueError, TypeError, IndexError): # 如果转换失败或索引越界返回默认值 return default elif isinstance(current, dict): # 如果当前层级是字典正常按键获取 current current[key] else: # 如果当前层级既不是字典也不是列表例如是字符串或数字无法继续深入 return default 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: 安全地在嵌套字典中设置值。 参数: data: 要修改的字典 path: 点分路径或键名列表 value: 要设置的值 separator: 用于分割路径字符串的分隔符 create_missing: 是否创建缺失的中间字典节点 返回: 如果设置成功返回 True否则返回 False # 将路径字符串分割为键名列表 if isinstance(path, str): keys path.split(separator) else: keys path # 如果路径为空无法设置 if not keys: return False current data # 遍历到最后一个键的父节点 for key in keys[:-1]: # 如果当前节点不存在 if key not in current: if create_missing: # 如果允许创建缺失节点则创建一个新的空字典 current[key] {} else: # 如果不允许创建则设置失败 return False # 移动到下一层 next_node current[key] # 检查下一层是否是字典因为只有字典才能包含子键 # 注意这里不支持在列表中自动插入元素只支持字典嵌套 if not isinstance(next_node, dict): # 如果中间节点存在但不是字典例如是列表或基本类型无法继续深入设置 return False current next_node # 设置最终的值 try: final_key keys[-1] # 如果最后一层是列表尝试作为索引设置可选功能这里简化为仅支持字典键 # 为了保持简单和一致性safe_set 主要针对字典结构。 # 如果用户希望设置列表元素通常需要先确保列表存在且长度足够这比较复杂。 # 此处我们假设最终目标是一个字典的键。 current[final_key] value return True except Exception: return Falsesafe_set会在需要时创建嵌套结构并设置值。这在动态构建字典时很有用。我们来测试这两个函数if __name__ __main__: # 示例嵌套数据 user_data { user: { name: Anna, address: { city: San Francisco, zip: 94105 }, orders: [ {id: 1, total: 99.99}, {id: 2, total: 149.50} ] } } print(--- Safe Get 测试 ---) # 1. 获取存在的城市 city safe_get(user_data, user.address.city) print(f城市: {city}) # 2. 获取不存在的国家使用默认值 country safe_get(user_data, user.address.country, defaultUnknown) print(f国家: {country}) # 3. 获取列表中的第一个订单总额 # 注意路径中的 0 会被识别为列表索引 first_order_total safe_get(user_data, user.orders.0.total) print(f第一笔订单总额: ${first_order_total}) # 4. 获取深层不存在的路径 missing safe_get(user_data, user.profile.bio, defaultNo bio) print(f个人简介: {missing}) # 5. 尝试在非字典节点上继续获取 (orders 是列表不能直接用 .id 获取所有 ID只能索引) invalid_path safe_get(user_data, user.orders.id, defaultInvalid Path) print(f无效路径结果: {invalid_path}) print(\n--- Safe Set 测试 ---) # 1. 在空字典中创建深层结构 new_data {} success safe_set(new_data, user.settings.theme, dark) print(f设置主题成功: {success}) print(f新数据结构: {new_data}) # 2. 修改已存在的值 success safe_set(user_data, user.address.city, New York) print(f修改城市成功: {success}) print(f新城市: {safe_get(user_data, user.address.city)}) # 3. 尝试在不允许创建缺失节点的情况下设置 temp_data {a: 1} success safe_set(temp_data, a.b.c, value, create_missingFalse) print(f在不允许创建缺失节点时设置深层路径成功: {success}) # 应为 False因为 a 是整数不是字典 # 4. 覆盖非字典节点 overwrite_data {x: {y: old}} # 先将 y 设置为一个字典以便后续可以在其下设置 z safe_set(overwrite_data, x.y, {z: new}) success safe_set(overwrite_data, x.y.z, final_value) print(f覆盖后设置深层值成功: {success}) print(f最终值: {safe_get(overwrite_data, x.y.z)})输出--- Safe Get 测试 --- 城市: San Francisco 国家: Unknown 第一笔订单总额: $99.99 个人简介: No bio 无效路径结果: Invalid Path --- Safe Set 测试 --- 设置主题成功: True 新数据结构: {user: {settings: {theme: dark}}} 修改城市成功: True 新城市: New York 在不允许创建缺失节点时设置深层路径成功: False 覆盖后设置深层值成功: True 最终值: final_value这种模式消除了防御性编程的杂乱代码使你在处理 JSON、配置文件或任何深层嵌套数据时代码更加整洁。为长时间运行的操作强制设置超时有些操作耗时过长。数据库查询可能挂起网页抓取可能卡在慢速服务器上或者计算可能永远运行下去。你需要一种方法来设定时间限制并及时退出。下面是一个使用线程的超时装饰器import threading import functools import time from typing import Callable, Optional, Any class TimeoutError(Exception): 当操作超过指定超时时限后抛出的异常。 pass def timeout(seconds: float, error_message: Optional[str] None): 一个用于强制函数执行超时的装饰器。 注意此实现使用线程。由于 Python 的 GIL 和线程机制 它无法真正“杀死”正在运行的线程只能停止等待并抛出异常。 后台线程可能会继续运行直到完成但其结果将被忽略。 参数: seconds: 最大执行时间秒 error_message: 超时时的自定义错误消息 返回: 装饰器函数 if seconds 0: raise ValueError(超时时间不能为负数) def decorator(func: Callable) - Callable: functools.wraps(func) def wrapper(*args, **kwargs) - Any: # 使用列表来存储结果或异常因为列表是可变对象可以在闭包中修改 # 初始化为 None表示尚未完成 result_container [None] exception_container [None] def target(): try: # 执行目标函数 result_container[0] func(*args, **kwargs) except Exception as e: # 捕获函数内部抛出的任何异常 exception_container[0] e # 创建并启动线程 thread threading.Thread(targettarget) thread.daemon True # 设置为守护线程主程序退出时自动结束 thread.start() # 等待线程完成最多等待 seconds 秒 thread.join(timeoutseconds) # 检查线程是否仍在运行即是否超时 if thread.is_alive(): # 如果超时抛出 TimeoutError # 注意线程实际上仍在后台运行但我们不再关心其结果 raise TimeoutError( error_message or f操作在 {seconds} 秒后超时 ) # 如果线程已结束检查是否有异常 if exception_container[0] is not None: raise exception_container[0] # 返回正常结果 return result_container[0] return wrapper return decorator这个装饰器在单独的线程中运行你的函数并使用thread.join(timeoutseconds)来等待。如果超时后线程仍然存活我们就知道它耗时过长于是抛出TimeoutError。函数结果存储在一个列表可变容器中以便内部线程能够修改它。如果线程中发生了异常我们会在主线程中重新抛出。⚠️ 一个限制即使超时后线程仍会在后台继续运行。对于大多数用例来说这没问题但对于有副作用的操作需要格外小心。我们来测试一下if __name__ __main__: print(--- 测试 1: 模拟慢速数据库查询 (应超时) ---) timeout(2, error_message查询耗时过长) def slow_database_query(): 模拟一个慢速查询。 time.sleep(5) return 查询结果 try: result slow_database_query() print(f结果: {result}) except TimeoutError as e: print(f超时错误: {e}) print(\n--- 测试 2: 模拟快速数据获取 (应成功) ---) timeout(3) def fetch_data(): 模拟一个快速操作。 time.sleep(1) return {data: value} try: data fetch_data() print(f成功: {data}) except TimeoutError as e: print(f超时错误: {e}) print(\n--- 测试 3: 函数内部抛出异常 ---) timeout(5) def faulty_function(): 模拟一个会抛出异常的函数。 raise ValueError(内部错误) try: faulty_function() except ValueError as e: print(f捕获到内部异常: {e}) except TimeoutError as e: print(f超时错误: {e}) print(\n--- 测试 4: 刚好在超时前完成 ---) timeout(2) def just_in_time(): 模拟一个刚好在超时前完成的操作。 time.sleep(1.9) return 及时完成 try: result just_in_time() print(f结果: {result}) except TimeoutError as e: print(f超时错误: {e})输出--- 测试 1: 模拟慢速数据库查询 (应超时) --- 超时错误: 查询耗时过长 --- 测试 2: 模拟快速数据获取 (应成功) --- 成功: {data: value} --- 测试 3: 函数内部抛出异常 --- 捕获到内部异常: 内部错误 --- 测试 4: 刚好在超时前完成 --- 结果: 及时完成这种模式对于构建响应式应用至关重要。在抓取网站、调用外部 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 ): 用于自动获取和清理资源的上下文管理器。 参数: acquire: 获取资源的函数 release: 释放资源的函数 on_error: 可选的错误处理函数接收异常对象和资源对象 suppress_errors: 是否在清理后抑制异常即不重新抛出 注意: 如果发生异常执行顺序为: . 调用 on_error (如果提供且资源已获取) . 如果 suppress_errors 为 False则重新抛出原始异常 . 始终执行 release (在 finally 块中) resource None try: # 获取资源 resource acquire() # 将资源 yield 给 with 块使用 yield resource except Exception as e: # 如果在 with 块中发生异常 if on_error and resource is not None: try: # 调用自定义错误处理器 on_error(e, resource) except Exception as handler_error: # 如果错误处理器本身出错打印警告但不掩盖原始异常 print(f错误处理器执行出错: {handler_error}) traceback.print_exc() # 根据配置决定是否抑制异常 if not suppress_errors: raise finally: # 无论是否发生异常都尝试释放资源 if resource is not None: try: release(resource) except Exception as cleanup_error: # 如果清理过程中出错打印错误信息 print(f资源清理过程中出错: {cleanup_error}) traceback.print_exc()managed_resource函数是一个上下文管理器工厂。它接受两个必需函数一个用于获取资源一个用于释放资源。释放函数始终在finally块中运行从而保证即使发生异常也会执行清理。可选的on_error参数让你可以在异常传播之前处理它。这对于日志记录、发送警报或尝试恢复很有用。suppress_errors标志决定异常是否被显式抛出还是被抑制。下面是一个辅助类用于演示资源操作# --- 辅助类资源追踪器 --- class ResourceTracker: 用于追踪资源操作的辅助类。 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(正在获取资源) return self def release(self): 模拟释放资源。 self.log(正在释放资源) def use(self, action: str): 模拟使用资源执行操作。 self.log(f使用资源执行: {action})我们来测试这个上下文管理器# --- 测试示例 --- if __name__ __main__: print(--- 测试 1: 带错误处理的正常异常流程 ---) tracker1 ResourceTracker(Database) def error_handler(exception, resource): resource.log(f捕获到错误: {exception}) resource.log(尝试执行回滚操作) try: with managed_resource( acquirelambda: tracker1.acquire(), releaselambda r: r.release(), on_errorerror_handler ) as db: db.use(INSERT INTO users) raise ValueError(重复条目错误) except ValueError as e: print(f主程序捕获到异常: {e}) print(\n--- 测试 2: 成功执行无异常 ---) tracker2 ResourceTracker(FileHandle) try: with managed_resource( acquirelambda: tracker2.acquire(), releaselambda r: r.release() ) as file: file.use(Write data) print(操作成功完成) except Exception as e: print(f意外异常: {e}) print(\n--- 测试 3: 抑制异常 ---) tracker3 ResourceTracker(NetworkSocket) try: with managed_resource( acquirelambda: tracker3.acquire(), releaselambda r: r.release(), suppress_errorsTrue ) as sock: sock.use(Send packet) raise ConnectionError(连接丢失) except Exception as e: print(f此异常应被抑制不应打印: {e}) else: print(异常已被抑制程序继续执行) print(\n--- 测试 4: 清理过程中出错 ---) tracker4 ResourceTracker(BadResource) def bad_release(r): r.log(尝试释放... 失败!) raise RuntimeError(释放失败) try: with managed_resource( acquirelambda: tracker4.acquire(), releasebad_release ) as res: res.use(Do something) except Exception as e: print(f主程序未捕获到清理异常因为它在 finally 中被捕获并打印: {type(e).__name__})输出--- 测试 1: 带错误处理的正常异常流程 --- [Database] 正在获取资源 [Database] 使用资源执行: INSERT INTO users [Database] 捕获到错误: 重复条目错误 [Database] 尝试执行回滚操作 [Database] 正在释放资源 主程序捕获到异常: 重复条目错误 --- 测试 2: 成功执行无异常 --- [FileHandle] 正在获取资源 [FileHandle] 使用资源执行: Write data [FileHandle] 正在释放资源 操作成功完成 --- 测试 3: 抑制异常 --- [NetworkSocket] 正在获取资源 [NetworkSocket] 使用资源执行: Send packet [NetworkSocket] 正在释放资源 异常已被抑制程序继续执行 --- 测试 4: 清理过程中出错 --- [BadResource] 正在获取资源 [BadResource] 使用资源执行: Do something [BadResource] 尝试释放... 失败! 资源清理过程中出错: 释放失败 Traceback (most recent call last): File ..., line ..., in managed_resource release(resource) File ..., line ..., in bad_release raise RuntimeError(释放失败) RuntimeError: 释放失败 主程序未捕获到清理异常因为它在 finally 中被捕获并打印: Exception这种模式对于管理数据库连接、文件句柄、网络套接字、锁以及任何需要保证清理的资源都非常有用。它可以防止资源泄漏使代码更加安全。总结文中的每个函数都针对一个特定的错误处理挑战重试临时性故障系统化验证输入安全访问嵌套数据防止操作挂起管理资源清理这些模式在 API 集成、数据处理管道、网页抓取和面向用户的应用中反复出现。文中使用的技术包括装饰器、上下文管理器和可组合函数使错误处理不再重复繁琐更加可靠。你可以将这些函数直接放入项目中使用也可以根据具体需求进行调整。它们独立、易懂能解决你日常会遇到的问题。