异常处理

一、什么是异常

异常是程序运行过程中出现的错误。

例如：

number = int("abc")

运行时会出现：

ValueError: invalid literal for int() with base 10: 'abc'

再例如：

result = 10 / 0

会出现：

ZeroDivisionError: division by zero

异常和语法错误不同。

语法错误：

if age >= 18
    print("成年")

程序根本无法正常开始执行。

异常：

number = int("abc")

语法没有问题，但运行到这一行时发生错误。

异常处理的作用是：

发现错误
→ 捕获错误
→ 记录错误
→ 给出提示
→ 决定继续、重试或终止

二、为什么需要异常处理

没有异常处理：

number = int(input("请输入整数："))

print(number * 2)

如果用户输入：

abc

程序会立即中断。

使用异常处理：

try:
    number = int(
        input("请输入整数：")
    )
except ValueError:
    print("请输入合法整数")
else:
    print(number * 2)

此时程序可以友好地处理错误。

异常处理常用于：

用户输入
文件读写
网络请求
JSON 解析
数据库操作
数值计算
类型转换
第三方接口
并发任务

三、try 和 except

基本格式：

try:
    可能发生异常的代码
except 异常类型:
    异常处理代码

示例：

try:
    number = int("abc")
except ValueError:
    print("转换失败")

运行结果：

转换失败

程序不会因为 ValueError 直接终止。

1. 执行流程

try:
    number = int("100")
except ValueError:
    print("转换失败")

print("程序继续运行")

因为没有发生异常，except 不会执行。

结果：

程序继续运行

如果发生异常：

try:
    number = int("abc")
except ValueError:
    print("转换失败")

print("程序继续运行")

结果：

转换失败
程序继续运行

四、获取异常信息

可以使用 as 获取异常对象：

try:
    number = int("abc")
except ValueError as error:
    print(f"转换失败：{error}")

结果类似：

转换失败：invalid literal for int() with base 10: 'abc'

异常对象中包含错误原因。

五、捕获多个异常

1. 分别处理

try:
    a = int(
        input("请输入第一个数字：")
    )
    b = int(
        input("请输入第二个数字：")
    )

    result = a / b
except ValueError:
    print("输入内容必须是数字")
except ZeroDivisionError:
    print("除数不能为零")

不同异常可以给出不同提示。

2. 同时捕获多个异常

如果多个异常使用相同处理逻辑，可以写成元组：

try:
    a = int(input("请输入数字："))
    result = 100 / a
except (
    ValueError,
    ZeroDivisionError,
) as error:
    print(f"计算失败：{error}")

必须使用圆括号：

except (ValueError, ZeroDivisionError):

六、except 匹配顺序

异常会按照 except 从上到下匹配。

try:
    number = int("abc")
except ValueError:
    print("值错误")
except Exception:
    print("其他错误")

输出：

值错误

子类异常应该放在前面，父类异常放在后面。

正确：

try:
    ...
except FileNotFoundError:
    ...
except OSError:
    ...
except Exception:
    ...

不推荐：

try:
    ...
except Exception:
    ...
except FileNotFoundError:
    ...

因为 FileNotFoundError 已经被 Exception 捕获，后面的分支永远不会执行。

七、else

else 在 try 没有发生异常时执行。

try:
    number = int(
        input("请输入整数：")
    )
except ValueError:
    print("输入格式错误")
else:
    print(
        f"输入成功：{number}"
    )

执行顺序：

try 成功
→ else

如果 try 发生异常：

try 失败
→ except

1. 为什么使用 else

下面的写法：

try:
    number = int("100")
    result = number * 2
    print(result)
except ValueError:
    print("转换失败")

try 中包含的代码太多。

更推荐：

try:
    number = int("100")
except ValueError:
    print("转换失败")
else:
    result = number * 2
    print(result)

原则：

try 中只放真正可能抛出目标异常的代码。

这样可以避免意外捕获其他错误。

八、finally

finally 中的代码无论是否发生异常都会执行。

try:
    number = int("100")
except ValueError:
    print("转换失败")
finally:
    print("程序处理结束")

结果：

程序处理结束

即使发生异常：

try:
    number = int("abc")
except ValueError:
    print("转换失败")
finally:
    print("程序处理结束")

结果：

转换失败
程序处理结束

1. finally 适用场景

finally 常用于释放资源：

关闭文件
关闭数据库连接
释放锁
清理临时文件
恢复程序状态

示例：

file = None

try:
    file = open(
        "data.txt",
        "r",
        encoding="utf-8",
    )

    content = file.read()
except FileNotFoundError:
    print("文件不存在")
finally:
    if file is not None:
        file.close()

不过文件操作更推荐使用 with：

with open(
    "data.txt",
    "r",
    encoding="utf-8",
) as file:
    content = file.read()

九、完整结构

完整异常处理结构：

try:
    可能发生异常的代码
except 某个异常:
    处理异常
else:
    没有异常时执行
finally:
    无论如何都会执行

示例：

try:
    number = int(
        input("请输入整数：")
    )
except ValueError as error:
    print(f"输入错误：{error}")
else:
    print(
        f"结果：{number * 2}"
    )
finally:
    print("本次操作结束")

十、常见内置异常

1. ValueError

值的格式或范围不正确：

int("abc")

float("hello")

2. TypeError

数据类型不正确：

"年龄：" + 25

len(100)

3. ZeroDivisionError

除数为零：

10 / 0

4. IndexError

列表索引超出范围：

numbers = [1, 2, 3]

print(numbers[10])

5. KeyError

字典键不存在：

stock = {
    "code": "600519",
}

print(stock["name"])

6. AttributeError

对象没有指定属性或方法：

number = 100

number.append(1)

7. NameError

变量不存在：

print(user_name)

8. FileNotFoundError

文件不存在：

open("missing.txt")

9. PermissionError

没有文件或目录访问权限：

open(
    "/protected/file.txt",
    "w",
)

10. OSError

操作系统相关错误。

例如：

文件系统错误
路径错误
设备错误
权限错误

FileNotFoundError 和 PermissionError 都属于 OSError 的子类。

11. ImportError

模块导入失败：

from module import missing_name

12. ModuleNotFoundError

模块不存在：

import missing_module

13. JSONDecodeError

JSON 格式错误：

import json

json.loads("{invalid json}")

它属于：

json.JSONDecodeError

14. TimeoutError

操作超时。

常见于：

网络请求
异步任务
文件系统
并发操作

15. RuntimeError

运行状态不符合要求，但没有更具体的异常类型。

16. NotImplementedError

表示子类或调用方必须实现某个功能：

class Animal:
    def speak(self) -> None:
        raise NotImplementedError

17. StopIteration

迭代器没有更多元素：

iterator = iter([1])

print(next(iterator))
print(next(iterator))

第二次调用会抛出 StopIteration。

通常由 for 循环自动处理。

十一、异常继承体系

大部分日常异常都继承自：

Exception

简化结构：

BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
    ├── ArithmeticError
    │   └── ZeroDivisionError
    ├── LookupError
    │   ├── IndexError
    │   └── KeyError
    ├── OSError
    │   ├── FileNotFoundError
    │   └── PermissionError
    ├── RuntimeError
    ├── TypeError
    └── ValueError

通常应该捕获：

Exception

而不是：

BaseException

因为 BaseException 还包含：

KeyboardInterrupt
SystemExit
GeneratorExit

捕获它们可能导致程序无法正常退出。

十二、不要使用裸 except

不推荐：

try:
    result = dangerous_operation()
except:
    print("出错了")

裸 except 会捕获几乎所有异常，包括用户按下 Ctrl+C 产生的 KeyboardInterrupt。

推荐捕获明确异常：

try:
    result = dangerous_operation()
except ValueError as error:
    print(error)

必要时捕获普通异常：

try:
    result = dangerous_operation()
except Exception as error:
    print(error)

十三、不要静默吞掉异常

不推荐：

try:
    result = calculate()
except Exception:
    pass

这会让错误完全消失，程序可能继续使用错误状态运行。

更合理：

try:
    result = calculate()
except ValueError as error:
    print(f"计算失败：{error}")

或者记录日志：

import logging

logger = logging.getLogger(__name__)

try:
    result = calculate()
except ValueError:
    logger.exception("计算失败")

十四、主动抛出异常

使用 raise 主动抛出异常。

def divide(
    a: float,
    b: float,
) -> float:
    if b == 0:
        raise ValueError(
            "除数不能为零"
        )

    return a / b

调用：

result = divide(10, 0)

会出现：

ValueError: 除数不能为零

1. 参数校验

def set_age(age: int) -> None:
    if not isinstance(age, int):
        raise TypeError(
            "age 必须是整数"
        )

    if age < 0:
        raise ValueError(
            "年龄不能小于零"
        )

    print(f"年龄：{age}")

类型错误使用：

TypeError

值不合法使用：

ValueError

2. 股票代码校验

def validate_stock_code(
    code: str,
) -> str:
    if not isinstance(code, str):
        raise TypeError(
            "股票代码必须是字符串"
        )

    code = code.strip()

    if len(code) != 6:
        raise ValueError(
            "股票代码长度必须为 6 位"
        )

    if not code.isdigit():
        raise ValueError(
            "股票代码必须全部由数字组成"
        )

    return code

调用：

try:
    code = validate_stock_code(
        " 600519 "
    )
except (
    TypeError,
    ValueError,
) as error:
    print(error)
else:
    print(code)

十五、重新抛出异常

在 except 中单独使用 raise，可以重新抛出当前异常。

def load_data() -> str:
    try:
        with open(
            "data.txt",
            "r",
            encoding="utf-8",
        ) as file:
            return file.read()
    except FileNotFoundError:
        print("记录错误信息")
        raise

调用者仍然可以继续处理：

try:
    content = load_data()
except FileNotFoundError:
    print("上层处理文件缺失")

单独使用：

raise

会保留原始异常和堆栈。

十六、自定义异常

当内置异常不能清楚表达业务含义时，可以创建自定义异常。

class StockError(Exception):
    """股票业务异常。"""

子异常：

class InvalidStockCodeError(
    StockError
):
    """股票代码无效。"""


class InvalidPriceError(
    StockError
):
    """股票价格无效。"""

使用：

def validate_stock_code(
    code: str,
) -> str:
    code = code.strip()

    if (
        len(code) != 6
        or not code.isdigit()
    ):
        raise InvalidStockCodeError(
            f"无效股票代码：{code}"
        )

    return code

捕获：

try:
    code = validate_stock_code(
        "ABC"
    )
except InvalidStockCodeError as error:
    print(error)

1. 自定义异常继承关系

class StockError(Exception):
    pass


class InvalidStockCodeError(
    StockError
):
    pass


class InvalidPriceError(
    StockError
):
    pass

可以分别捕获：

try:
    ...
except InvalidStockCodeError:
    ...
except InvalidPriceError:
    ...

也可以统一捕获：

try:
    ...
except StockError:
    ...

2. 自定义异常保存数据

class InvalidStockCodeError(
    ValueError
):
    def __init__(
        self,
        code: str,
        reason: str,
    ) -> None:
        self.code = code
        self.reason = reason

        super().__init__(
            f"股票代码 {code!r} 无效："
            f"{reason}"
        )

使用：

raise InvalidStockCodeError(
    code="ABC",
    reason="必须是 6 位数字",
)

捕获后可以访问：

try:
    ...
except InvalidStockCodeError as error:
    print(error.code)
    print(error.reason)

十七、异常链

底层异常转换为业务异常时，应该保留原始原因。

class InvalidPriceError(
    ValueError
):
    pass

def parse_price(
    value: str,
) -> float:
    try:
        return float(value)
    except ValueError as error:
        raise InvalidPriceError(
            f"价格格式无效：{value!r}"
        ) from error

调用：

parse_price("abc")

异常信息会显示：

原始 ValueError
由它导致的 InvalidPriceError

from error 明确说明两个异常之间的因果关系。

1. 隐藏底层异常

有时不希望把内部实现暴露给调用者：

def parse_price(
    value: str,
) -> float:
    try:
        return float(value)
    except ValueError:
        raise InvalidPriceError(
            "价格格式无效"
        ) from None

此时只显示新的业务异常。

使用原则：

需要调试底层原因 → from error
不希望暴露内部细节 → from None

十八、文件异常处理

from pathlib import Path

def read_text_file(
    file_path: Path,
) -> str:
    try:
        return file_path.read_text(
            encoding="utf-8"
        )
    except FileNotFoundError as error:
        raise FileNotFoundError(
            f"文件不存在：{file_path}"
        ) from error
    except PermissionError as error:
        raise PermissionError(
            f"无权读取文件：{file_path}"
        ) from error
    except UnicodeDecodeError as error:
        raise ValueError(
            f"文件编码不是 UTF-8："
            f"{file_path}"
        ) from error

调用：

try:
    content = read_text_file(
        Path("data.txt")
    )
except (
    FileNotFoundError,
    PermissionError,
    ValueError,
) as error:
    print(error)

十九、JSON 异常处理

import json
from pathlib import Path

def load_json(
    file_path: Path,
) -> object:
    try:
        text = file_path.read_text(
            encoding="utf-8"
        )
    except FileNotFoundError as error:
        raise FileNotFoundError(
            f"JSON 文件不存在："
            f"{file_path}"
        ) from error

    try:
        return json.loads(text)
    except json.JSONDecodeError as error:
        raise ValueError(
            "JSON 格式错误："
            f"第 {error.lineno} 行，"
            f"第 {error.colno} 列"
        ) from error

JSONDecodeError 常用属性：

error.msg
error.lineno
error.colno
error.pos

二十、字典数据校验

def parse_stock(
    data: dict[str, object],
) -> dict[str, object]:
    try:
        code = str(data["code"])
        name = str(data["name"])
        price = float(data["price"])
    except KeyError as error:
        raise ValueError(
            f"缺少字段：{error.args[0]}"
        ) from error
    except (
        TypeError,
        ValueError,
    ) as error:
        raise ValueError(
            "股票数据类型错误"
        ) from error

    return {
        "code": code,
        "name": name,
        "price": price,
    }

调用：

data = {
    "code": "600519",
    "name": "贵州茅台",
}

try:
    stock = parse_stock(data)
except ValueError as error:
    print(error)

结果：

缺少字段：price

二十一、循环中的异常处理

有一组数据需要逐条处理：

values = [
    "10",
    "20",
    "abc",
    "30",
]

1. 单条失败不影响其他数据

results: list[int] = []

for value in values:
    try:
        number = int(value)
    except ValueError:
        print(
            f"跳过无效数据：{value}"
        )
        continue

    results.append(number)

print(results)

结果：

[10, 20, 30]

2. 记录错误数据

results: list[int] = []
errors: list[
    tuple[str, str]
] = []

for value in values:
    try:
        number = int(value)
    except ValueError as error:
        errors.append(
            (
                value,
                str(error),
            )
        )
    else:
        results.append(number)

3. 是否应该跳过错误

取决于业务要求。

可以跳过：

日志中的个别坏数据
批量清洗中的少数异常行
可容忍的第三方接口失败

不应该直接跳过：

财务核心数据
交易指令
数据库事务
身份验证
关键配置文件

二十二、异常和返回值

不推荐使用特殊返回值代表错误：

def divide(
    a: float,
    b: float,
) -> float | None:
    if b == 0:
        return None

    return a / b

调用者可能忘记判断：

result = divide(10, 0)

print(result * 2)

更推荐抛出异常：

def divide(
    a: float,
    b: float,
) -> float:
    if b == 0:
        raise ZeroDivisionError(
            "除数不能为零"
        )

    return a / b

调用者必须明确处理：

try:
    result = divide(10, 0)
except ZeroDivisionError as error:
    print(error)

1. 什么时候返回 None

找不到数据通常可以返回 None：

def find_stock(
    code: str,
) -> dict[str, object] | None:
    for stock in stocks:
        if stock["code"] == code:
            return stock

    return None

因为“没有找到”可能是正常情况。

2. 什么时候抛出异常

以下情况更适合抛出异常：

参数类型错误
数据格式错误
必要字段缺失
文件无法读取
数据库操作失败
业务规则被违反

例如：

def buy_stock(
    balance: float,
    amount: float,
) -> float:
    if amount <= 0:
        raise ValueError(
            "买入金额必须大于零"
        )

    if amount > balance:
        raise ValueError(
            "资金不足"
        )

    return balance - amount

二十三、日志记录异常

正式项目中应使用 logging。

import logging

logger = logging.getLogger(__name__)

普通错误日志：

try:
    number = int("abc")
except ValueError as error:
    logger.error(
        "数字转换失败：%s",
        error,
    )

记录完整异常堆栈：

try:
    number = int("abc")
except ValueError:
    logger.exception(
        "数字转换失败"
    )

logger.exception() 应放在 except 中使用。

它会自动记录：

异常类型
异常信息
调用堆栈
出错位置

二十四、断言 assert

assert 用于检查程序内部假设。

def calculate_average(
    numbers: list[float],
) -> float:
    assert numbers, (
        "numbers 不应该为空"
    )

    return sum(numbers) / len(numbers)

断言失败：

AssertionError

1. assert 不适合业务校验

不推荐：

def set_price(
    price: float,
) -> None:
    assert price >= 0

因为运行 Python 时可以通过优化选项关闭断言：

python -O script.py

业务校验应该使用：

def set_price(
    price: float,
) -> None:
    if price < 0:
        raise ValueError(
            "价格不能小于零"
        )

assert 更适合：

开发期检查
内部不变量
测试
理论上不应该发生的状态

二十五、异常组 ExceptionGroup

Python 3.11 以上支持异常组。

异常组可以同时保存多个异常。

errors = [
    ValueError("价格错误"),
    TypeError("类型错误"),
]

raise ExceptionGroup(
    "批量处理失败",
    errors,
)

1. 为什么需要异常组

批量处理多条数据时，可能同时发现多个错误：

records = [
    {
        "code": "600519",
        "price": 1500,
    },
    {
        "code": "ABC",
        "price": 100,
    },
    {
        "code": "000001",
        "price": -10,
    },
]

可以一次收集所有错误：

def validate_records(
    records: list[
        dict[str, object]
    ],
) -> None:
    errors: list[Exception] = []

    for index, record in enumerate(
        records,
        start=1,
    ):
        code = str(
            record.get("code", "")
        )

        price = record.get("price")

        if (
            len(code) != 6
            or not code.isdigit()
        ):
            errors.append(
                ValueError(
                    f"第 {index} 条代码无效"
                )
            )

        if not isinstance(
            price,
            (int, float),
        ):
            errors.append(
                TypeError(
                    f"第 {index} 条价格类型错误"
                )
            )
        elif price < 0:
            errors.append(
                ValueError(
                    f"第 {index} 条价格不能小于零"
                )
            )

    if errors:
        raise ExceptionGroup(
            "数据校验失败",
            errors,
        )

2. except*

使用 except* 分别捕获异常组中的异常类型：

try:
    validate_records(records)
except* TypeError as error:
    print(
        "类型错误：",
        error,
    )
except* ValueError as error:
    print(
        "值错误：",
        error,
    )

普通 except 不能像 except* 那样拆分异常组中的不同类型。

二十六、并发任务中的异常

使用线程池：

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
)

def process_code(
    code: str,
) -> str:
    if not code.isdigit():
        raise ValueError(
            f"无效代码：{code}"
        )

    return code

codes = [
    "600519",
    "ABC",
    "000001",
]

with ThreadPoolExecutor() as executor:
    future_map = {
        executor.submit(
            process_code,
            code,
        ): code
        for code in codes
    }

    for future in as_completed(
        future_map
    ):
        code = future_map[future]

        try:
            result = future.result()
        except ValueError as error:
            print(
                f"{code} 处理失败："
                f"{error}"
            )
        except Exception:
            logger.exception(
                "%s 出现未知错误",
                code,
            )
        else:
            print(
                f"{code} 处理成功："
                f"{result}"
            )

并发任务中的异常通常在：

future.result()

时重新抛出。

二十七、异步异常处理

import asyncio

async def fetch_data(
    code: str,
) -> str:
    await asyncio.sleep(0.1)

    if code == "ABC":
        raise ValueError(
            "无效股票代码"
        )

    return f"{code} 数据"

普通捕获：

async def main() -> None:
    try:
        result = await fetch_data(
            "ABC"
        )
    except ValueError as error:
        print(error)
    else:
        print(result)


asyncio.run(main())

1. asyncio.gather

默认情况下，一个协程失败会向上抛出异常：

async def main() -> None:
    results = await asyncio.gather(
        fetch_data("600519"),
        fetch_data("ABC"),
        fetch_data("000001"),
    )

    print(results)

也可以让异常作为结果返回：

async def main() -> None:
    results = await asyncio.gather(
        fetch_data("600519"),
        fetch_data("ABC"),
        fetch_data("000001"),
        return_exceptions=True,
    )

    for result in results:
        if isinstance(
            result,
            Exception,
        ):
            print(
                f"任务失败：{result}"
            )
        else:
            print(
                f"任务成功：{result}"
            )

使用 return_exceptions=True 时，不能忘记检查结果中的异常对象。

2. TaskGroup

Python 3.11 以上支持：

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(
                fetch_data("600519")
            )

            group.create_task(
                fetch_data("ABC")
            )
    except* ValueError as error:
        print(
            f"任务组失败：{error}"
        )

TaskGroup 中多个任务失败时，可能产生异常组。

二十八、上下文管理器中的异常

自定义上下文管理器：

class Resource:
    def __enter__(self):
        print("获取资源")
        return self

    def __exit__(
        self,
        exc_type,
        exc_value,
        traceback,
    ) -> bool:
        print("释放资源")

        if exc_type is not None:
            print(
                f"发生异常："
                f"{exc_value}"
            )

        return False

使用：

with Resource():
    raise ValueError(
        "测试错误"
    )

__exit__() 返回：

False

表示异常继续向外传播。

如果返回：

True

表示异常已处理，不再传播。

通常不建议随意吞掉异常。

二十九、contextlib.suppress

如果某种异常确实可以安全忽略，可以使用：

from contextlib import suppress
from pathlib import Path

file_path = Path("temp.txt")

with suppress(
    FileNotFoundError
):
    file_path.unlink()

等价于：

try:
    file_path.unlink()
except FileNotFoundError:
    pass

适合：

删除可能不存在的临时文件
清理可选资源
明确允许失败的操作

不适合：

with suppress(Exception):
    important_operation()

这会隐藏所有问题。

三十、数据库事务中的异常

伪代码示例：

connection = get_connection()

try:
    connection.begin()

    update_balance()
    create_order()

except DatabaseError:
    connection.rollback()
    raise
else:
    connection.commit()
finally:
    connection.close()

处理原则：

成功 → commit
失败 → rollback
最后 → close

很多数据库库支持上下文管理器：

with connection:
    execute_operations()

具体行为取决于数据库驱动。

三十一、重试机制

某些临时错误可以重试，例如：

网络短暂失败
接口超时
服务暂时不可用

from time import sleep

def run_with_retry(
    operation,
    *,
    max_attempts: int = 3,
    delay: float = 1.0,
):
    if max_attempts <= 0:
        raise ValueError(
            "max_attempts 必须大于零"
        )

    last_error: Exception | None = None

    for attempt in range(
        1,
        max_attempts + 1,
    ):
        try:
            return operation()
        except TimeoutError as error:
            last_error = error

            if attempt == max_attempts:
                break

            sleep(delay)

    raise RuntimeError(
        f"操作重试 "
        f"{max_attempts} 次后仍然失败"
    ) from last_error

只重试临时性异常。

不要重试：

参数错误
权限错误
数据格式错误
业务规则错误

三十二、批量数据处理案例

定义异常：

class StockDataError(Exception):
    pass


class MissingFieldError(
    StockDataError
):
    pass


class InvalidCodeError(
    StockDataError
):
    pass


class InvalidPriceError(
    StockDataError
):
    pass

解析单条股票：

def parse_stock(
    data: dict[str, object],
) -> dict[str, object]:
    try:
        raw_code = data["code"]
        raw_name = data["name"]
        raw_price = data["price"]
    except KeyError as error:
        raise MissingFieldError(
            f"缺少字段：{error.args[0]}"
        ) from error

    code = str(raw_code).strip()
    name = str(raw_name).strip()

    if (
        len(code) != 6
        or not code.isdigit()
    ):
        raise InvalidCodeError(
            f"无效股票代码：{code!r}"
        )

    try:
        price = float(raw_price)
    except (
        TypeError,
        ValueError,
    ) as error:
        raise InvalidPriceError(
            f"价格格式无效："
            f"{raw_price!r}"
        ) from error

    if price < 0:
        raise InvalidPriceError(
            "价格不能小于零"
        )

    return {
        "code": code,
        "name": name,
        "price": price,
    }

批量处理：

def parse_stocks(
    records: list[
        dict[str, object]
    ],
) -> tuple[
    list[dict[str, object]],
    list[tuple[int, Exception]],
]:
    valid_records: list[
        dict[str, object]
    ] = []

    errors: list[
        tuple[int, Exception]
    ] = []

    for index, record in enumerate(
        records,
        start=1,
    ):
        try:
            stock = parse_stock(
                record
            )
        except StockDataError as error:
            errors.append(
                (
                    index,
                    error,
                )
            )
        else:
            valid_records.append(
                stock
            )

    return valid_records, errors

使用：

records = [
    {
        "code": "600519",
        "name": "贵州茅台",
        "price": 1500,
    },
    {
        "code": "ABC",
        "name": "错误数据",
        "price": 10,
    },
    {
        "code": "000001",
        "name": "平安银行",
        "price": "11.5",
    },
]

valid_records, errors = (
    parse_stocks(records)
)

print(valid_records)

for index, error in errors:
    print(
        f"第 {index} 条失败："
        f"{error}"
    )

三十三、异常测试

使用 unittest：

import unittest

class TestValidateStockCode(
    unittest.TestCase
):
    def test_valid_code(
        self,
    ) -> None:
        self.assertEqual(
            validate_stock_code(
                "600519"
            ),
            "600519",
        )

    def test_invalid_length(
        self,
    ) -> None:
        with self.assertRaises(
            ValueError
        ):
            validate_stock_code(
                "60051"
            )

    def test_invalid_type(
        self,
    ) -> None:
        with self.assertRaises(
            TypeError
        ):
            validate_stock_code(
                600519
            )

测试异常信息：

with self.assertRaisesRegex(
    ValueError,
    "长度必须为 6 位",
):
    validate_stock_code(
        "60051"
    )

三十四、异常处理常见错误

1. 捕获范围过大

不推荐：

try:
    load_data()
    clean_data()
    save_data()
except Exception:
    print("失败")

无法判断是哪一步失败。

更推荐缩小范围：

try:
    data = load_data()
except FileNotFoundError as error:
    ...

try:
    cleaned_data = clean_data(data)
except ValueError as error:
    ...

2. 捕获异常后什么都不做

不推荐：

try:
    process()
except Exception:
    pass

至少应该：

记录日志
返回明确状态
转换为业务异常
重新抛出

3. 捕获了无法处理的异常

如果当前层不知道如何解决，不要强行处理。

try:
    save_to_database(data)
except DatabaseError:
    raise

或者根本不捕获，让上层处理。

4. 使用 Exception 代替具体异常

不推荐：

try:
    number = int(value)
except Exception:
    ...

5. 用异常代替正常流程

不推荐：

try:
    value = data["name"]
except KeyError:
    value = "未知"

简单默认值可以使用：

value = data.get(
    "name",
    "未知",
)

但如果字段必须存在，则应该保留 KeyError 或转换为业务异常。

6. 错误地返回 finally

危险写法：

def test():
    try:
        return 1
    finally:
        return 2

结果是：

finally 中的 return 会覆盖 try 中的返回值。

更严重的是，它还可能吞掉异常：

def test():
    try:
        raise ValueError(
            "错误"
        )
    finally:
        return 2

函数会返回 2，异常消失。

原则：

不要在 finally 中使用 return、break 或 continue。

7. 修改异常后丢失原因

不推荐：

try:
    float(value)
except ValueError:
    raise InvalidPriceError(
        "价格错误"
    )

推荐明确异常链：

try:
    float(value)
except ValueError as error:
    raise InvalidPriceError(
        "价格错误"
    ) from error

三十五、异常处理设计原则

原则一：在能够处理的层级捕获

底层函数：

def load_file(
    file_path: Path,
) -> str:
    return file_path.read_text(
        encoding="utf-8"
    )

业务层：

def load_config(
    file_path: Path,
) -> dict[str, object]:
    try:
        text = load_file(file_path)
    except FileNotFoundError as error:
        raise ConfigurationError(
            "配置文件不存在"
        ) from error

界面层：

try:
    config = load_config(path)
except ConfigurationError as error:
    print(
        f"程序无法启动：{error}"
    )

不同层级负责不同事情：

底层 → 抛出技术异常
业务层 → 转换为业务异常
界面层 → 给用户提示

原则二：只捕获能够处理的异常

能够处理：

except FileNotFoundError:
    create_default_file()

不能处理：

except Exception:
    print("不知道怎么办")

不能处理时，让异常继续传播通常更合理。

原则三：异常信息要具体

不推荐：

raise ValueError(
    "数据错误"
)

原则四：不要暴露敏感信息

错误信息中不要包含：

数据库密码
API 密钥
用户密码
完整身份信息
内部服务器凭证

不推荐：

raise ConnectionError(
    f"连接失败，密码为："
    f"{password}"
)

原则五：异常应该表达异常情况

正常业务结果不一定是异常。

例如查询不到可选数据：

return None

但必要数据不存在：

raise RecordNotFoundError

需要根据业务语义决定。

三十六、推荐模板

输入转换

try:
    number = int(value)
except ValueError as error:
    raise ValueError(
        f"无法转换为整数："
        f"{value!r}"
    ) from error

文件读取

try:
    content = file_path.read_text(
        encoding="utf-8"
    )
except FileNotFoundError as error:
    raise FileNotFoundError(
        f"文件不存在：{file_path}"
    ) from error
except PermissionError as error:
    raise PermissionError(
        f"没有读取权限：{file_path}"
    ) from error

JSON 解析

try:
    data = json.loads(text)
except json.JSONDecodeError as error:
    raise ValueError(
        "JSON 格式错误，"
        f"位置："
        f"{error.lineno}:"
        f"{error.colno}"
    ) from error

批量处理

for item in items:
    try:
        result = process(item)
    except ExpectedError as error:
        logger.warning(
            "跳过数据 %r：%s",
            item,
            error,
        )
        continue

    results.append(result)

顶层程序入口

def main() -> int:
    try:
        run_application()
    except ConfigurationError as error:
        logger.error(
            "配置错误：%s",
            error,
        )
        return 1
    except KeyboardInterrupt:
        logger.info(
            "用户中断程序"
        )
        return 130
    except Exception:
        logger.exception(
            "程序发生未处理异常"
        )
        return 1

    return 0

if __name__ == "__main__":
    raise SystemExit(main())

三十七、练习题

练习一

编写函数，将字符串转换为整数。如果转换失败，抛出带中文提示的 ValueError。

参考答案：

def parse_integer(
    value: str,
) -> int:
    try:
        return int(value)
    except ValueError as error:
        raise ValueError(
            f"无法转换为整数："
            f"{value!r}"
        ) from error

练习二

编写除法函数，除数为零时抛出异常。

参考答案：

def divide(
    a: float,
    b: float,
) -> float:
    if b == 0:
        raise ZeroDivisionError(
            "除数不能为零"
        )

    return a / b

练习三

读取文本文件，分别处理文件不存在和没有权限。

参考答案：

from pathlib import Path


def read_file(
    file_path: Path,
) -> str:
    try:
        return file_path.read_text(
            encoding="utf-8"
        )
    except FileNotFoundError as error:
        raise FileNotFoundError(
            f"文件不存在：{file_path}"
        ) from error
    except PermissionError as error:
        raise PermissionError(
            f"无权读取：{file_path}"
        ) from error

练习四

定义自定义异常 InvalidAgeError，年龄小于零或大于 150 时抛出。

参考答案：

class InvalidAgeError(
    ValueError
):
    pass

def validate_age(
    age: int,
) -> int:
    if not isinstance(age, int):
        raise TypeError(
            "年龄必须是整数"
        )

    if not 0 <= age <= 150:
        raise InvalidAgeError(
            f"年龄超出有效范围："
            f"{age}"
        )

    return age

练习五

批量转换字符串列表，将合法整数和错误信息分别保存。

参考答案：

def parse_numbers(
    values: list[str],
) -> tuple[
    list[int],
    list[tuple[str, str]],
]:
    numbers: list[int] = []
    errors: list[
        tuple[str, str]
    ] = []

    for value in values:
        try:
            number = int(value)
        except ValueError as error:
            errors.append(
                (
                    value,
                    str(error),
                )
            )
        else:
            numbers.append(number)

    return numbers, errors

练习六

解析 JSON 文件，错误时给出行号和列号。

参考答案：

import json
from pathlib import Path


def read_json(
    file_path: Path,
) -> object:
    text = file_path.read_text(
        encoding="utf-8"
    )

    try:
        return json.loads(text)
    except json.JSONDecodeError as error:
        raise ValueError(
            f"JSON 格式错误："
            f"第 {error.lineno} 行，"
            f"第 {error.colno} 列"
        ) from error

三十八、异常处理核心总结

基本捕获：

try:
    operation()
except ValueError as error:
    print(error)

多个异常：

except (
    ValueError,
    TypeError,
) as error:
    ...

成功时执行：

else:
    ...

始终执行：

finally:
    ...

主动抛出：

raise ValueError(
    "参数错误"
)

重新抛出：

except ValueError:
    raise

异常链：

raise BusinessError(
    "业务失败"
) from error

自定义异常：

class BusinessError(
    Exception
):
    pass

异常组：

raise ExceptionGroup(
    "多个错误",
    errors,
)

分别捕获：

except* ValueError:
    ...

异常处理的核心原则：

捕获明确异常
缩小 try 范围
只处理能够处理的异常
必要时重新抛出
保留异常原因
不要静默吞掉错误
不要在 finally 中 return
错误信息要具体
关键错误需要记录日志
业务异常应有清晰类型

异常处理不是让错误消失，而是：

让错误被发现、被解释、被记录，并以可控方式处理。

一个成熟的程序不是永远不出错，而是在出错时仍然能够保持清晰、稳定和可追踪。