我要评论错误处理往往是优秀代码中最薄弱的环节。
像缺失主键、请求失败、函数运行时间过长等问题在实际项目中频繁出现。
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_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,))
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", 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 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(target=target)
thread.daemon = true # 设置为守护线程，主程序退出时自动结束
thread.start()
# 等待线程完成，最多等待 seconds 秒
thread.join(timeout=seconds)
# 检查线程是否仍在运行（即是否超时）
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(timeout=seconds) 来等待。如果超时后线程仍然存活，我们就知道它耗时过长，于是抛出 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(
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")
def bad_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 集成、数据处理管道、网页抓取和面向用户的应用中反复出现。
文中使用的技术包括装饰器、上下文管理器和可组合函数，使错误处理不再重复繁琐，更加可靠。
你可以将这些函数直接放入项目中使用，也可以根据具体需求进行调整。
它们独立、易懂，能解决你日常会遇到的问题。
到此这篇关于5 个实用的python错误处理函数的文章就介绍到这了,更多相关python错误处理函数内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论