展开菜单
首页 精品内容 本月促销 装机必备 Windows PDF教程 macOS软件 IOS软件 Android AI 专题
全部分类

当前位置:

首页 > 编程开发 > 5 个实用的Python错误处理函数详解

5 个实用的Python错误处理函数详解

错误处理往往是优秀代码中最薄弱的环节。 像缺失主键、请求失败、函数运行时间过长这些问题,在实际项目中几乎天天都能碰到。 Python 自带的 try-except 块确实有用,但光靠它还真不够——很多真实场景根本覆盖不到。 更好的思路是:把常见的失败情形封装成小巧、可复用的函数,比如带限制的重试、输

本文目录

    错误处理往往是优秀代码中最薄弱的环节。

    5 个实用的Python错误处理函数详解

    像缺失主键、请求失败、函数运行时间过长这些问题,在实际项目中几乎天天都能碰到。

    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 时,延迟时间依次为1秒、2秒、4秒、8秒——给压力下的系统留出恢复时间。

    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 False

    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 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 集成、数据处理管道、网页抓取和面向用户的应用中反复出现。

    文中使用的技术包括装饰器、上下文管理器和可组合函数,让错误处理不再重复繁琐,更加可靠。

    你可以将这些函数直接放入项目中使用,也可以根据具体需求调整。

    它们独立、易懂,能解决日常工作中切实遇到的麻烦。

    本站声明:本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系bd@zhengruan.com
    作者最新文章
    编程开发 Python
    上一篇: 蓝云怎么使用
    相关文章 更多
    精品专题 更多
    装机必备

    正软商城装机必备专区,精选办公、浏览器、安全防护、影音播放、压缩解压、设计创作和系统工具等电脑常用正版软件,帮助用户快速完成新电脑软件配置。

    Windows

    正软商城Windows软件专区,汇集适用于Windows电脑的办公、设计、安全防护、影音播放、开发工具和系统优化软件,提供软件介绍、系统要求、正版授权及购买下载服务。

    PDF教程

    正软商城PDF教程频道提供PDF编辑、转换、合并、拆分、压缩及格式处理方法,同时介绍常用PDF软件和工具的使用技巧。

    macOS软件

    正软商城macOS软件专区,精选适用于Mac电脑的办公、设计、影音、效率、开发和系统工具,提供软件功能介绍、macOS兼容版本、正版授权及购买下载服务。

    IOS软件

    正软商城iOS软件专区,精选适用于iPhone和iPad的办公、学习、影音、设计、效率及AI应用,提供功能介绍、适用设备、系统要求和正版获取方式等信息。

    AI

    正软商城AI软件专区,汇集AI写作、AI绘画、AI视频、AI办公、AI编程、AI翻译、智能客服和数据分析等人工智能工具,提供功能介绍、适用平台、收费方式及正版购买信息。

    Mac软件 更多
    Aperty macOS版
    Aperty macOS版

    Aperty 是 Skylum 推出的 AI 人像修图软件,支持自然肤质修饰、瑕疵处理、面部与身体调整、预设及批量同步。

    火山引擎 网页版
    火山引擎 网页版

    火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力,依托字节自研豆包多模态大模型,凭借火山方舟一站式平台,提供模型精调、推理部署、智能应用搭建等全链路服务,结合高性能AI算力,助力各行业快速落地大模型数字化与智能化应用。

    立刻MV 网页版
    立刻MV 网页版

    立刻MV 面向音乐人、内容创作者和品牌团队,帮助用户低成本完成歌曲 MV、歌词视频、AI 分镜、图生视频与最终合成。

    蛙蛙写作 网页版
    蛙蛙写作 网页版

    蛙蛙写作 AI 创作工具,覆盖小说、剧本、文案,一键生成文稿,打通文字转视频全流程,高效解决创作难题。

    讯飞智作 网页版
    讯飞智作 网页版

    讯飞智作是一款基于语音合成技术、虚拟人驱动技术与数字人形象声音构建能力的 AI 数字人音视频创作平台。

    超级简历WonderCV 网页版
    超级简历WonderCV 网页版

    超级简历WonderCV是面向应届生和AI小白的智能求职平台,覆盖简历、面试与职业规划。

    音述AI 网页版
    音述AI 网页版

    音述AI,是一个以“用声音述说故事”为核心的 AI 音乐创作与声音分享社区。

    AionClaw macOS版
    AionClaw macOS版

    AionClaw 是可在桌面端运行的任务型 AI 助手,通过数字专家与技能市场组合办公、创作、数据和开发能力。

    Loomy macOS版
    Loomy macOS版

    Loomy 是科大讯飞推出的桌面 AI 助手,可通过对话、任务与工作目录协助处理办公、内容创作和资料整理。

    AirBuddy 3 macOS版
    AirBuddy 3 macOS版

    AirBuddy 3 是面向 macOS 的蓝牙外设增强工具,可集中查看 AirPods、Beats 与苹果设备电量,提供快速连接、播放控制、电量提醒、桌面小组件和多台 Mac 外设接力等功能。

    AirServer macOS版
    AirServer macOS版

    AirServer macOS版可把 Mac 转换为无线投屏接收端,通过 AirPlay 或 Google Cast 接收手机、平板和电脑画面,并支持多设备同时连接、高清镜像及会话录制,适合家庭娱乐、课堂教学、会议演示和应用展示。

    AnyDesk macOS版
    AnyDesk macOS版

    AnyDesk macOS版是一款适用于远程办公、设备维护和客户支持的远程桌面工具,可连接并操作异地计算机,支持文件传输、无人值守访问、会话权限控制及跨平台协作。

    WINDOWS 更多
    讯飞绘文 网页版
    讯飞绘文 网页版

    讯飞绘文网页版是科大讯飞推出的AI图文创作与内容运营工具,覆盖热点选题、文章生成、稿件校对、智能配图、排版发布和数据复盘等环节,适合自媒体创作者及内容团队在线使用。

    Maya(Autodesk Maya) Windows版
    Maya(Autodesk Maya) Windows版

    Autodesk Maya 是面向影视、电视、动画和游戏行业的专业三维制作软件,集多边形与 NURBS 建模、角色绑定、关键帧和动作捕捉动画、Bifrost 特效模拟、材质开发及 Arnold 渲染于一体,

    Aperty Windows版
    Aperty Windows版

    Aperty 是 Skylum 推出的 AI 人像修图软件,支持自然肤质修饰、瑕疵处理、面部与身体调整、预设及批量同步。

    火山引擎 网页版
    火山引擎 网页版

    火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力,依托字节自研豆包多模态大模型,凭借火山方舟一站式平台,提供模型精调、推理部署、智能应用搭建等全链路服务,结合高性能AI算力,助力各行业快速落地大模型数字化与智能化应用。

    立刻MV 网页版
    立刻MV 网页版

    立刻MV 面向音乐人、内容创作者和品牌团队,帮助用户低成本完成歌曲 MV、歌词视频、AI 分镜、图生视频与最终合成。

    蛙蛙写作 网页版
    蛙蛙写作 网页版

    蛙蛙写作 AI 创作工具,覆盖小说、剧本、文案,一键生成文稿,打通文字转视频全流程,高效解决创作难题。

    讯飞智作 网页版
    讯飞智作 网页版

    讯飞智作是一款基于语音合成技术、虚拟人驱动技术与数字人形象声音构建能力的 AI 数字人音视频创作平台。

    超级简历WonderCV 网页版
    超级简历WonderCV 网页版

    超级简历WonderCV是面向应届生和AI小白的智能求职平台,覆盖简历、面试与职业规划。

    音述AI 网页版
    音述AI 网页版

    音述AI,是一个以“用声音述说故事”为核心的 AI 音乐创作与声音分享社区。

    AionClaw Windows版
    AionClaw Windows版

    AionClaw 是可在桌面端运行的任务型 AI 助手,通过数字专家与技能市场组合办公、创作、数据和开发能力。

    Loomy Windows版
    Loomy Windows版

    Loomy 是科大讯飞推出的桌面 AI 助手,可通过对话、任务与工作目录协助处理办公、内容创作和资料整理。

    AirServer Windows版
    AirServer Windows版

    AirServer Windows版可把电脑变成AirPlay、Google Cast和Miracast无线投屏接收端,适合会议演示、课堂教学及家庭大屏分享。发送设备通常无需另装客户端,并可同时接收和智能排列多个镜像会话。