错误处理往往是优秀代码中最薄弱的环节。
像缺失主键、请求失败、函数运行时间过长等问题在实际项目中频繁出现。
Python 自带的 try-except 块虽然有用，但单靠它并不能覆盖许多实际场景。
你需要将常见的失败情形封装成小型、可复用的函数，以支持带限制的重试、输入验证，以及防止代码运行超时的保护机制。
本文介绍的五个错误处理函数，可用于网页抓取、API 构建、用户数据处理等任务。
使用指数退避重试失败操作
在许多项目中，API 调用和网络请求经常失败。初学者的做法是只尝试一次，捕获异常并记录，然后停止。更好的做法是重试。这时指数退避就派上了用场。与其立即对失效服务发起多次重试（这只会让情况更糟），不如在每次尝试之间等待更长的时间：1 秒、2 秒、4 秒，以此类推。
我们来构建一个实现此功能的装饰器：
import timeimport functoolsfrom typing import Callable, Type, Tupledefretry_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: 需要捕获并重试的异常类型元组 """defdecorator(func: Callable): @functools.wraps(func)defwrapper(*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 wrapperreturn decorator
这个装饰器会包装你的函数，并捕获指定的异常。关键计算是 delay = base_delay * (exponential_base ** attempt)。当 base_delay=1、exponential_base=2 时，延迟时间为 1s、2s、4s、8s。这给了压力下的系统恢复的时间。
exceptions 参数让你可以指定哪些错误需要重试。例如，你可以重试 ConnectionError 但不重试 ValueError，因为连接问题是暂时的，而验证错误不是。
下面来看它的实际使用：
if __name__ == "__main__": counter = 0 @retry_with_backoff(max_attempts=3, base_delay=0.5, exceptions=(ValueError,))defunstable_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, OptionalclassValidationError(Exception):"""当验证失败时抛出的异常。"""def__init__(self, field: str, errors: List[str]): self.field = field self.errors = errors# 调用父类构造函数，设置异常消息为字段名和错误列表的组合 super().__init__(f"{field}: {', '.join(errors)}")
这个异常包含多个错误信息。当验证失败时，我们希望向用户展示所有的问题，而不仅仅是第一个错误。
下面是验证器本身：
defvalidate_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 isNone: messages = {} errors = []# 遍历所有验证规则for rule_name, rule_func in rules.items():try:# 执行验证函数ifnot 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 的函数。这使得规则可组合、可复用。
我们来创建一些常用的验证规则：
# --- 可复用的验证规则工厂函数 ---defnot_empty(value: Any) -> bool:"""检查值是否非空（针对字符串去除空白后）"""if isinstance(value, str):return bool(value.strip())return bool(value)defmin_length(min_len: int) -> Callable[[Any], bool]:"""生成一个检查最小长度的验证函数"""def_validator(value: Any) -> bool:try:return len(str(value)) >= min_lenexcept Exception:returnFalsereturn _validatordefmax_length(max_len: int) -> Callable[[Any], bool]:"""生成一个检查最大长度的验证函数"""def_validator(value: Any) -> bool:try:return len(str(value)) <= max_lenexcept Exception:returnFalsereturn _validatordefin_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_valexcept (ValueError, TypeError):# 如果无法转换为浮点数，则验证失败returnFalsereturn _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, Uniondefsafe_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 = datafor key in keys:try:# 处理列表索引：如果当前层级是列表，尝试将键转换为整数索引if isinstance(current, list):try: index = int(key) current = current[index]except (ValueError, TypeError, IndexError):# 如果转换失败或索引越界，返回默认值return defaultelif isinstance(current, dict):# 如果当前层级是字典，正常按键获取 current = current[key]else:# 如果当前层级既不是字典也不是列表（例如是字符串或数字），无法继续深入return defaultexcept (KeyError, IndexError, TypeError):# 捕获键错误、索引错误或类型错误，返回默认值return defaultreturn current
该函数将路径拆分为单个键，然后逐步遍历嵌套结构。如果某个键不存在，或者试图对不可下标对象进行下标操作，它会返回默认值而不是崩溃。它还能自动处理列表索引——如果当前值是列表且键为数字，则会将键转换为整数。
配套的设置函数如下：
defsafe_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# 如果路径为空，无法设置ifnot keys:returnFalse current = data# 遍历到最后一个键的父节点for key in keys[:-1]:# 如果当前节点不存在if key notin current:if create_missing:# 如果允许创建缺失节点，则创建一个新的空字典 current[key] = {}else:# 如果不允许创建，则设置失败returnFalse# 移动到下一层 next_node = current[key]# 检查下一层是否是字典，因为只有字典才能包含子键# 注意：这里不支持在列表中自动插入元素，只支持字典嵌套ifnot isinstance(next_node, dict):# 如果中间节点存在但不是字典（例如是列表或基本类型），无法继续深入设置returnFalse current = next_node# 设置最终的值try: final_key = keys[-1]# 如果最后一层是列表，尝试作为索引设置（可选功能，这里简化为仅支持字典键）# 为了保持简单和一致性，safe_set 主要针对字典结构。# 如果用户希望设置列表元素，通常需要先确保列表存在且长度足够，这比较复杂。# 此处我们假设最终目标是一个字典的键。 current[final_key] = valuereturnTrueexcept Exception:returnFalse
safe_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", default="Unknown") 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", default="No bio") print(f"个人简介: {missing}")# 5. 尝试在非字典节点上继续获取 (orders 是列表，不能直接用 .id 获取所有 ID，只能索引) invalid_path = safe_get(user_data, "user.orders.id", default="Invalid 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_missing=False) 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 threadingimport functoolsimport timefrom typing import Callable, Optional, AnyclassTimeoutError(Exception):"""当操作超过指定超时时限后抛出的异常。"""passdeftimeout(seconds: float, error_message: Optional[str] = None):""" 一个用于强制函数执行超时的装饰器。 注意：此实现使用线程。由于 Python 的 GIL 和线程机制， 它无法真正“杀死”正在运行的线程，只能停止等待并抛出异常。 后台线程可能会继续运行直到完成，但其结果将被忽略。 参数: seconds: 最大执行时间（秒） error_message: 超时时的自定义错误消息 返回: 装饰器函数 """if seconds < 0:raise ValueError("超时时间不能为负数")defdecorator(func: Callable) -> Callable: @functools.wraps(func)defwrapper(*args, **kwargs) -> Any:# 使用列表来存储结果或异常，因为列表是可变对象，可以在闭包中修改# 初始化为 None，表示尚未完成 result_container = [None] exception_container = [None]deftarget():try:# 执行目标函数 result_container[0] = func(*args, **kwargs)except Exception as e:# 捕获函数内部抛出的任何异常 exception_container[0] = e# 创建并启动线程 thread = threading.Thread(target=target) thread.daemon = True# 设置为守护线程，主程序退出时自动结束 thread.start()# 等待线程完成，最多等待 seconds 秒 thread.join(timeout=seconds)# 检查线程是否仍在运行（即是否超时）if thread.is_alive():# 如果超时，抛出 TimeoutError# 注意：线程实际上仍在后台运行，但我们不再关心其结果raise TimeoutError( error_message orf"操作在 {seconds} 秒后超时" )# 如果线程已结束，检查是否有异常if exception_container[0] isnotNone:raise exception_container[0]# 返回正常结果return result_container[0]return wrapperreturn decorator
这个装饰器在单独的线程中运行你的函数，并使用 thread.join(timeout=seconds) 来等待。如果超时后线程仍然存活，我们就知道它耗时过长，于是抛出 TimeoutError。函数结果存储在一个列表（可变容器）中，以便内部线程能够修改它。如果线程中发生了异常，我们会在主线程中重新抛出。
⚠️ 一个限制：即使超时后，线程仍会在后台继续运行。对于大多数用例来说这没问题，但对于有副作用的操作，需要格外小心。
我们来测试一下：
if __name__ == "__main__": print("--- 测试 1: 模拟慢速数据库查询 (应超时) ---") @timeout(2, error_message="查询耗时过长")defslow_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)deffetch_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)deffaulty_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)defjust_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 contextmanagerfrom typing import Callable, Any, Optionalimport traceback@contextmanagerdefmanaged_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 = Nonetry:# 获取资源 resource = acquire()# 将资源 yield 给 with 块使用yield resourceexcept Exception as e:# 如果在 with 块中发生异常if on_error and resource isnotNone:try:# 调用自定义错误处理器 on_error(e, resource)except Exception as handler_error:# 如果错误处理器本身出错，打印警告，但不掩盖原始异常 print(f"错误处理器执行出错: {handler_error}") traceback.print_exc()# 根据配置决定是否抑制异常ifnot suppress_errors:raisefinally:# 无论是否发生异常，都尝试释放资源if resource isnotNone:try: release(resource)except Exception as cleanup_error:# 如果清理过程中出错，打印错误信息 print(f"资源清理过程中出错: {cleanup_error}") traceback.print_exc()
managed_resource 函数是一个上下文管理器工厂。它接受两个必需函数：一个用于获取资源，一个用于释放资源。释放函数始终在 finally 块中运行，从而保证即使发生异常也会执行清理。
可选的 on_error 参数让你可以在异常传播之前处理它。这对于日志记录、发送警报或尝试恢复很有用。suppress_errors 标志决定异常是否被显式抛出还是被抑制。
下面是一个辅助类，用于演示资源操作：
# --- 辅助类：资源追踪器 ---classResourceTracker:"""用于追踪资源操作的辅助类。"""def__init__(self, name: str, verbose: bool = True): self.name = name self.verbose = verbose self.operations = []deflog(self, operation: str):"""记录操作并可选地打印。""" self.operations.append(operation)if self.verbose: print(f"[{self.name}] {operation}")defacquire(self):"""模拟获取资源。""" self.log("正在获取资源")return selfdefrelease(self):"""模拟释放资源。""" self.log("正在释放资源")defuse(self, action: str):"""模拟使用资源执行操作。""" self.log(f"使用资源执行: {action}")
我们来测试这个上下文管理器：
# --- 测试示例 ---if __name__ == "__main__": print("--- 测试 1: 带错误处理的正常异常流程 ---") tracker1 = ResourceTracker("Database")deferror_handler(exception, resource): resource.log(f"捕获到错误: {exception}") resource.log("尝试执行回滚操作")try:with managed_resource( acquire=lambda: tracker1.acquire(), release=lambda r: r.release(), on_error=error_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( acquire=lambda: tracker2.acquire(), release=lambda 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( acquire=lambda: tracker3.acquire(), release=lambda r: r.release(), suppress_errors=True ) as sock: sock.use("Send packet")raise ConnectionError("连接丢失")except Exception as e: print(f"此异常应被抑制，不应打印: {e}")else: print("异常已被抑制，程序继续执行") print("\n--- 测试 4: 清理过程中出错 ---") tracker4 = ResourceTracker("BadResource")defbad_release(r): r.log("尝试释放... 失败!")raise RuntimeError("释放失败")try:with managed_resource( acquire=lambda: tracker4.acquire(), release=bad_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 集成、数据处理管道、网页抓取和面向用户的应用中反复出现。
文中使用的技术包括装饰器、上下文管理器和可组合函数，使错误处理不再重复繁琐，更加可靠。
你可以将这些函数直接放入项目中使用，也可以根据具体需求进行调整。
它们独立、易懂，能解决你日常会遇到的问题。
