Python：异常处理、上下文管理和资源清理应该怎么写才稳

发表于 分类于 本文字数： 3k 阅读时长 ≈ 11 分钟
异常处理、上下文管理和资源清理如果写得随意，脚本很快就会出现文件没关、临时数据没清、报错后状态不一致的问题。这篇文章围绕一个实际的批量通知脚本，把这三块内容串起来。

Python 脚本写到一定阶段后，最容易暴露出来的问题已经不是“会不会写功能”，而是“出错以后会不会收场”。

最常见的几类问题通常是：

  • 文件打开了，但异常一抛就没关
  • 临时目录创建了，但脚本中途失败没清掉
  • 一部分通知发出去了，失败名单却没落盘
  • 捕获异常时只会 except Exception，结果真正的问题被吞了

这几类问题单看都不像大问题，但一旦脚本要长期跑、定时跑、批量跑，就会越来越明显。

这一篇直接围绕一个实际脚本展开：做一个批量通知脚本。

这个脚本要完成这些事：

  1. 读取一批待通知用户
  2. 模拟发送通知
  3. 把失败名单写进日志文件
  4. 不管中间是否异常，都要把文件和临时资源收好

用这个场景把三块内容串起来：

  1. 异常处理负责让脚本知道哪里失败了
  2. 上下文管理负责让资源按范围自动回收
  3. 资源清理负责保证失败之后也不会留下烂状态

一、这篇文章要解决什么问题

读完这一篇，应该能独立完成这些动作：

  • 在真正需要的地方捕获异常，而不是整段代码一把包住
  • 用 with 管理文件和临时资源
  • 理解 try / except / else / finally 分别适合放什么
  • 给批量脚本补失败日志和清理动作
  • 排查几类典型错误，例如异常被吞、文件没关、清理逻辑没执行

如果这些动作能独立做出来，Python 脚本就不再只是“能跑完顺利路径”，而是开始具备处理失败路径的能力。

二、先看要完成的实际功能

假设现在有一份待通知用户列表：

1
2
3
4
5
users = [
    {"name": "Tom", "email": "tom@example.com"},
    {"name": "Lucy", "email": "lucy@example.com"},
    {"name": "Mike", "email": "bad-email"},
]

现在要写一个脚本，执行后输出类似这样：

1
2
3
4
5
6
7
8
发送成功: Tom
发送成功: Lucy
发送失败: Mike, error=邮箱格式不合法

总人数: 3
成功数量: 2
失败数量: 1
失败日志: notify_output/failed.log

这个场景不复杂，但已经足够把异常、上下文管理和清理动作接进来。

三、先从最小版本开始

先别急着写 with 和 finally，先把最基础的通知逻辑跑起来。

新建 notify_users.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def send_email(user):
    if "@" not in user["email"]:
        raise ValueError("邮箱格式不合法")

    print(f'发送成功: {user["name"]}')


users = [
    {"name": "Tom", "email": "tom@example.com"},
    {"name": "Lucy", "email": "lucy@example.com"},
    {"name": "Mike", "email": "bad-email"},
]

for user in users:
    send_email(user)

执行：

1
python notify_users.py

输出可能是：

1
2
3
4
5
发送成功: Tom
发送成功: Lucy
Traceback (most recent call last):
...
ValueError: 邮箱格式不合法

这个版本当然能看出问题，但一旦第三个人失败，整个脚本就停了，而且没有失败日志。

这就是异常处理最该上场的地方。

四、异常处理最实际的用法是什么

第一次写异常处理时，最容易走两个极端：

  • 完全不捕获，脚本一出错就退出
  • 什么都 except Exception，然后直接 pass

这两种都不稳。

实际做法是：在知道怎么处理错误的地方捕获，在不知道怎么处理的地方继续抛出。

1. 先给单次通知补最小异常处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def send_email(user):
    if "@" not in user["email"]:
        raise ValueError("邮箱格式不合法")

    print(f'发送成功: {user["name"]}')


users = [
    {"name": "Tom", "email": "tom@example.com"},
    {"name": "Lucy", "email": "lucy@example.com"},
    {"name": "Mike", "email": "bad-email"},
]

failed_users = []

for user in users:
    try:
        send_email(user)
    except ValueError as exc:
        print(f'发送失败: {user["name"]}, error={exc}')
        failed_users.append(user["name"])

输出：

1
2
3
发送成功: Tom
发送成功: Lucy
发送失败: Mike, error=邮箱格式不合法

这里关键点是：

  • 只捕获自己预期会出现的 ValueError
  • 失败时把用户名记下来
  • 脚本继续处理后面的用户

2. 一个实际错误：把所有异常都吞掉

错误写法：

1
2
3
4
5
for user in users:
    try:
        send_email(user)
    except Exception:
        pass

这种写法的问题很实际：

  • 出错了，但日志里什么都看不到
  • 后面统计为什么少一个人时，很难排查

修复至少要做到两点：

1
2
3
4
5
for user in users:
    try:
        send_email(user)
    except ValueError as exc:
        print(f'发送失败: {user["name"]}, error={exc}')

如果确实要兜底，也应该保留错误信息，而不是直接吞掉。

3. else 和 finally 应该放什么

看一个更完整的结构：

1
2
3
4
5
6
7
8
try:
    send_email(user)
except ValueError as exc:
    print(f"失败: {exc}")
else:
    print("这一条通知发送成功")
finally:
    print("这一轮发送结束")

这四块分工很明确：

  • try：放可能出错的代码
  • except：放失败时的处理逻辑
  • else：放没有异常时才执行的逻辑
  • finally：放不管成功失败都要执行的收尾动作

五、上下文管理为什么比手动开关资源更稳

1. 先看最常见的文件写入场景

最常见的失败名单写法通常是：

1
2
3
f = open("failed.log", "w", encoding="utf-8")
f.write("Mike\n")
f.close()

这当然能工作，但一旦中间抛异常，close() 很可能就走不到。

2. 实际写法：用 with

1
2
with open("failed.log", "w", encoding="utf-8") as f:
    f.write("Mike\n")

这里 with 最重要的价值不是语法简洁，而是：代码块结束后，不管中间有没有异常，资源都会按协议退出。

3. 一个实际错误：文件写一半异常，资源状态不清楚

错误写法：

1
2
3
4
f = open("failed.log", "w", encoding="utf-8")
f.write("Tom\n")
raise RuntimeError("中途失败")
f.close()

这里真正的问题不是只少了一个 close()，而是你已经没法保证：

  • 缓冲区是否完整刷到磁盘
  • 文件句柄是否还占着
  • 后续逻辑是否还能安全写同一个文件

实际写法：

1
2
3
with open("failed.log", "w", encoding="utf-8") as f:
    f.write("Tom\n")
    raise RuntimeError("中途失败")

即使异常抛出，文件资源也会退出。

六、资源清理不是只指关文件

在实际脚本里，资源清理通常还包括：

  • 临时目录
  • 临时文件
  • 锁文件
  • 网络连接
  • 数据库连接

这一篇先用最容易验证的临时目录来讲。

1. 手动清理的写法容易漏

例如：

1
2
3
4
5
6
7
8
9
10
11
from pathlib import Path

tmp_dir = Path("tmp_notify")
tmp_dir.mkdir(exist_ok=True)

cache_file = tmp_dir / "cache.txt"
cache_file.write_text("cached", encoding="utf-8")

# 中间如果报错，下面的清理就可能走不到
# cache_file.unlink()
# tmp_dir.rmdir()

2. 实际写法：交给上下文管理器

1
2
3
4
5
6
7
8
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as tmp_dir:
    tmp_path = Path(tmp_dir)
    cache_file = tmp_path / "cache.txt"
    cache_file.write_text("cached", encoding="utf-8")
    print(cache_file.exists())

输出：

1
True

代码块结束后，临时目录会自动清掉。

这就是资源清理最实际的一层：不用靠记忆保证清理动作，而是把清理绑定到代码块范围。

七、把异常处理、上下文管理和清理真正串起来

现在把三者放回批量通知脚本里。

notify_users.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from pathlib import Path


def send_email(user):
    if "@" not in user["email"]:
        raise ValueError("邮箱格式不合法")

    print(f'发送成功: {user["name"]}')


def notify_users(users, output_dir: Path):
    output_dir.mkdir(parents=True, exist_ok=True)
    failed_path = output_dir / "failed.log"

    success_count = 0
    failed_count = 0

    with failed_path.open("w", encoding="utf-8") as failed_file:
        for user in users:
            try:
                send_email(user)
            except ValueError as exc:
                print(f'发送失败: {user["name"]}, error={exc}')
                failed_file.write(f'{user["name"]},{exc}\n')
                failed_count += 1
            else:
                success_count += 1

    print(f"\n总人数: {len(users)}")
    print(f"成功数量: {success_count}")
    print(f"失败数量: {failed_count}")
    print(f"失败日志: {failed_path}")


def main():
    users = [
        {"name": "Tom", "email": "tom@example.com"},
        {"name": "Lucy", "email": "lucy@example.com"},
        {"name": "Mike", "email": "bad-email"},
    ]
    notify_users(users, Path("notify_output"))


if __name__ == "__main__":
    main()

执行：

1
python notify_users.py

输出：

1
2
3
4
5
6
7
8
发送成功: Tom
发送成功: Lucy
发送失败: Mike, error=邮箱格式不合法

总人数: 3
成功数量: 2
失败数量: 1
失败日志: notify_output/failed.log

八、再补一层：不管成败都要执行的清理动作

如果脚本中间还创建了临时缓存，就需要 finally 或上下文管理器一起配合。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import tempfile
from pathlib import Path


def run_notify_job(users):
    with tempfile.TemporaryDirectory() as tmp_dir:
        tmp_path = Path(tmp_dir)
        cache_file = tmp_path / "notify.cache"

        try:
            cache_file.write_text("start", encoding="utf-8")
            notify_users(users, Path("notify_output"))
        finally:
            print("本轮通知任务结束，临时目录将自动清理")

这里 finally 负责：

  • 打印收尾信息
  • 放必须执行的动作

而临时目录的实际删除交给上下文管理器。

九、完整版本放在一起

下面给出一个可以直接运行的完整版本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import tempfile
from pathlib import Path


def send_email(user):
    if "@" not in user["email"]:
        raise ValueError("邮箱格式不合法")

    print(f'发送成功: {user["name"]}')


def notify_users(users, output_dir: Path):
    output_dir.mkdir(parents=True, exist_ok=True)
    failed_path = output_dir / "failed.log"

    success_count = 0
    failed_count = 0

    with failed_path.open("w", encoding="utf-8") as failed_file:
        for user in users:
            try:
                send_email(user)
            except ValueError as exc:
                print(f'发送失败: {user["name"]}, error={exc}')
                failed_file.write(f'{user["name"]},{exc}\n')
                failed_count += 1
            else:
                success_count += 1

    print(f"\n总人数: {len(users)}")
    print(f"成功数量: {success_count}")
    print(f"失败数量: {failed_count}")
    print(f"失败日志: {failed_path}")


def run_notify_job(users):
    with tempfile.TemporaryDirectory() as tmp_dir:
        tmp_path = Path(tmp_dir)
        cache_file = tmp_path / "notify.cache"

        try:
            cache_file.write_text("start", encoding="utf-8")
            notify_users(users, Path("notify_output"))
        finally:
            print("本轮通知任务结束，临时目录将自动清理")


def main():
    users = [
        {"name": "Tom", "email": "tom@example.com"},
        {"name": "Lucy", "email": "lucy@example.com"},
        {"name": "Mike", "email": "bad-email"},
    ]
    run_notify_job(users)


if __name__ == "__main__":
    main()

十、怎么测试这一层是不是真的掌握了

这一层不能只看懂，要自己改一遍。

可以直接做这些动作：

  1. 增加一个会触发 RuntimeError 的场景
  2. 把失败名单同时写到屏幕和文件
  3. 改成按批次写多个失败日志
  4. 给 send_email() 补一个空邮箱校验
  5. 确认异常后临时目录确实被清掉

如果这些改动都能独立完成，说明异常处理、上下文管理和清理动作已经开始真正会用了。

十一、一个实际排错场景

这类脚本里一个非常常见的实际问题是：失败日志文件创建出来了，但内容是空的。

例如第一版可能这样写：

1
2
3
4
5
6
7
8
9
10
failed_file = open("failed.log", "w", encoding="utf-8")

for user in users:
    try:
        send_email(user)
    except ValueError:
        failed_file.write(user["name"] + "\n")
        raise

failed_file.close()

这里一旦第一次失败就重新抛异常，close() 很可能走不到。

这时排查顺序应该很直接：

  1. 先看文件是不是用 with 打开的
  2. 再看异常抛出后资源退出逻辑有没有执行
  3. 再看失败日志写入动作是不是放在正确分支

如果文件内容不完整，先看资源退出和异常流转，不要先怀疑编码或磁盘。

十二、一个实际练习

可以直接把这一篇变成一个完整练习。

练习目标：做一个“批量处理用户数据脚本”。

要求：

  1. 读取一批用户数据
  2. 对每条数据做校验
  3. 成功数据和失败数据分开记录
  4. 失败原因写入日志文件
  5. 使用 with 管理文件
  6. 使用临时目录或 finally 做收尾清理

如果这个练习能独立做完，说明这一篇最核心的失败路径处理已经开始真正掌握。

十三、这篇文章学完以后，下一步应该补什么

如果这一篇已经能跟着做完，下一步最适合继续补的是：

  1. 文件处理、序列化、时间处理和路径管理有哪些高频坑
  2. pytest 里怎样验证异常和失败路径
  3. 项目从脚本到工程时，日志、配置和任务入口怎么继续整理

因为到这一步，脚本已经不只是写功能，而是在处理“出错以后还能不能稳住”的问题了。

十四、结语

异常处理、上下文管理和资源清理最好不要拆成三块孤立知识去背。

在实际脚本里，它们通常是在解决同一件事：

  • 出错时知道怎么收集失败信息
  • 资源使用完后能按范围退出
  • 脚本中断后也不会留下脏状态

只要把它们放进一个实际功能里一起练，失败路径就不会再只是“出了错再说”，而会慢慢变成脚本的一部分。