第7章 7.3 异常处理与日志

上一章我们折腾了半天 RESTful API 设计，好不容易把接口写出来了。结果上线第一天，用户打电话来："程序崩了！报了一堆英文错误，看不懂！"你一脸懵，打开服务器日志一看——全是 Fatal error: Maximum execution time exceeded 之类的天书。这章我们就来解决这个问题：让你的程序不仅能干活，还要优雅地出错。

🎯 开场 3 分钟：为什么要学这个？

你有没有遇到过这种情况：

• 场景 1：你写了个读取文件的程序，文件被删了，程序直接闪退
• 场景 2：线上出了 bug，但日志只写了"出错了"三个字，你根本不知道哪错的
• 场景 3：调用第三方接口，超时了，你的程序直接崩溃，用户以为你产品有问题

痛点总结：程序出错不可怕，可怕的是出错后你什么都不知道。

这章学完，你能：
1. 让程序在出错时优雅地崩溃（而不是闪退）
2. 用日志记录一切，事后能还原案发现场
3. 写一个真实可用的错误追踪工具，以后调试效\n\n\n\n\n\n率翻倍

🧱 基础 25 分钟：核心概念

异常处理是什么？

类比时间：想象你在餐厅吃饭。

• 正常情况：你点菜 → 厨房做菜 → 服务员上菜 → 你吃饭
• 异常情况：你点菜 → 厨房说"排骨卖完了" → 服务员说"给您换成红烧肉可以吗" → 你说行 → 继续吃饭

如果没这套机制呢？你点排骨，厨房说没了，然后...就没有然后了，你饿着肚子回家。

程序也是一样。没有异常处理：程序遇到错误 → 直接崩溃 → 用户一脸懵。有异常处理：程序遇到错误 → 我们捕获它 → 决定怎么应对 → 程序继续运行（或优雅退出）。

try/except/finally 三兄弟

Python 的异常处理靠这三个关键字：

# 01_basic_exception.py
try:
# 尝试执行这段代码
result = 10 / 0  # 故意除以零，看会发生什么
print(f"结果是: {result}")
except ZeroDivisionError:
# 如果上面的代码抛出 ZeroDivisionError，执行这里
print("糟糕，除数不能为零！")
finally:
# 不管有没有出错，这段都会执行
print("无论如何都会执行到这里")

运行结果：

糟糕，除数不能为零！
无论如何都会执行到这里

解释：第 3 行故意除以零，第 4 行根本不会执行。except 捕获了 ZeroDivisionError 异常，我们打印了一条友好提示，而不是让程序崩溃。

捕获多种异常

# 02_multiple_exceptions.py
try:
# 模拟一个可能出错的操作
number = int(input("请输入一个数字: "))
result = 100 / number
print(f"100 除以 {number} 等于 {result}")
except ValueError:
# 用户输入了非数字
print("喂，我说的是数字！")
except ZeroDivisionError:
# 用户输入了 0
print("0 不能做除数，想让程序爆炸吗？")
except Exception as e:
# 兜底：其他所有未知错误
print(f"未知错误: {e}")
finally:
print("程序结束")

解释：
- ValueError：用户输入 "abc" 时触发
- ZeroDivisionError：用户输入 "0" 时触发
- Exception as e：前面都没捕获到的错误，用 e 拿到错误信息

自定义异常：自己定义"错法"

有时候内置的异常不够用，你想定义自己的错误类型。

类比：餐厅可以把错误细分成"菜卖完了""厨房着火""服务员辞职"——不同的错误类型，处理方式不同。

# 03_custom_exception.py

# 第1步：定义一个自定义异常类
class NotPositiveNumberError(Exception):
"""当传入非正数时抛出这个异常"""
def __init__(self, value):
    self.value = value
    super().__init__(f"数值必须为正数，你传的是: {value}")

# 第2步：写一个会抛出异常的函数
def divide_if_valid(a, b):
if b <= 0:
    raise NotPositiveNumberError(b)
return a / b

# 第3步：捕获并处理
try:
result = divide_if_valid(10, -3)
except NotPositiveNumberError as e:
print(f"捕获到自定义异常: {e}")

运行结果：

捕获到自定义异常: 数值必须为正数，你传的是: -3

解释：
- raise 关键字用于抛出异常
- 自定义异常继承自 Exception
- super().__init__() 调用父类构造函数，传入错误信息

日志系统：程序的"黑匣子"

类比：飞机失事后，调查人员会看黑匣子记录飞行数据。程序崩溃后，日志就是你的"黑匣子"。

为什么不用 print？
- print 只能输出到屏幕，程序关了就没了
- print 输出不了时间、级别等元信息
- 没法分类（哪些是错误，哪些是警告）

Python 内置了 logging 模块，用法很简单：

# 04_logging_basic.py
import logging

# 配置日志
logging.basicConfig(
level=logging.DEBUG,  # 记录所有级别
format='%(asctime)s - %(levelname)s - %(message)s',  # 格式
)

# 使用不同级别记录日志
logging.debug("这是调试信息")      # 排错用
logging.info("这是普通信息")       # 正常运行
logging.warning("这是警告")        # 可能有问题
logging.error("这是错误")          # 出错了

运行结果：

2024-01-15 10:30:45,123 - DEBUG - 这是调试信息
2024-01-15 10:30:45,124 - INFO - 这是普通信息
2024-01-15 10:30:45,125 - WARNING - 这是警告
2024-01-15 10:30:45,126 - ERROR - 这是错误

format 参数说明：
- %(asctime)s：时间
- %(levelname)s：日志级别
- %(message)s：日志内容

日志输出到文件

生产环境一定要输出到文件，而不是只打印到屏幕：

# 05_logging_to_file.py
import logging

logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='app.log',      # 输出到文件
filemode='a'             # 追加模式，不覆盖
)

logging.info("程序启动")
logging.info("用户登录: 张三")
logging.error("数据库连接失败")
logging.info("程序结束")

运行后，当前目录会生成 app.log 文件，内容类似：

2024-01-15 10:30:45 - INFO - 程序启动
2024-01-15 10:30:46 - INFO - 用户登录: 张三
2024-01-15 10:30:47 - ERROR - 数据库连接失败
2024-01-15 10:30:48 - INFO - 程序结束

完整的异常+日志组合拳

真实项目中，异常处理和日志是配合使用的：

# 06_exception_with_logging.py
import logging

logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
filename='error.log'
)

def read_config(filename):
"""读取配置文件"""
try:
    with open(filename, 'r', encoding='utf-8') as f:
        content = f.read()
        logging.info(f"成功读取配置文件: {filename}")
        return content
except FileNotFoundError:
    logging.error(f"配置文件不存在: {filename}")
    return None
except PermissionError:
    logging.error(f"没有权限读取文件: {filename}")
    return None
except Exception as e:
    logging.critical(f"读取配置文件时发生未知错误: {e}", exc_info=True)
    return None

# 测试
result = read_config('config.txt')
print(f"读取结果: {result}")

解释：
- exc_info=True：记录完整的堆栈跟踪，对调试很有用
- logging.critical：最高级别，用于严重错误

🔥 实战 35 分钟：3 个递进小项目

项目 1：带日志的文件处理工具（5 分钟）

需求：读取一个文件，统计字数，优雅处理文件不存在的情况。

# project_1_word_counter.py
import logging

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='word_counter.log'
)

def count_words(filename):
"""统计文件字数"""
try:
    with open(filename, 'r', encoding='utf-8') as f:
        content = f.read()
        word_count = len(content)
        logging.info(f"成功统计: {filename}，字数: {word_count}")
        return word_count
except FileNotFoundError:
    logging.warning(f"文件不存在: {filename}，返回 0")
    return 0
except Exception as e:
    logging.error(f"统计字数时出错: {e}")
    return 0

# 测试
if __name__ == "__main__":
# 先测试一个不存在的文件
count = count_words("不存在的文件.txt")
print(f"字数: {count}")

# 再测试一个存在的文件
with open("test.txt", "w", encoding="utf-8") as f:
    f.write("Hello World 你好世界")
count = count_words("test.txt")
print(f"字数: {count}")

预期输出：

字数: 0
字数: 12

一句话解释：文件不存在时返回 0 而不是崩溃，所有操作都被记录到日志。

项目 2：API 调用器，带超时和错误重试（15 分钟）

需求：调用一个外部 API（我们模拟），如果超时或出错就重试 3 次，记录每次尝试的结果。

# project_2_api_caller.py
import logging
import random
import time

# 配置日志
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='api_caller.log'
)

class APIError(Exception):
"""自定义 API 错误"""
pass

def simulate_api_call(endpoint, data):
"""模拟调用 API，成功率 50%"""
# 模拟网络请求
time.sleep(0.5)

# 随机模拟成功或失败
if random.random() < 0.5:
    logging.info(f"API 调用成功: {endpoint}")
    return {"status": "success", "data": data}
else:
    error_msg = random.choice([
        "Connection timeout",
        "500 Internal Server Error",
        "Rate limit exceeded"
    ])
    logging.warning(f"API 调用失败: {error_msg}")
    raise APIError(error_msg)

def call_api_with_retry(endpoint, data, max_retries=3, delay=1):
"""带重试的 API 调用"""
for attempt in range(1, max_retries + 1):
    logging.info(f"尝试 {attempt}/{max_retries}: 调用 {endpoint}")
    try:
        result = simulate_api_call(endpoint, data)
        logging.info(f"最终成功: {endpoint}")
        return result
    except APIError as e:
        logging.warning(f"第 {attempt} 次尝试失败: {e}")
        if attempt < max_retries:
            logging.info(f"等待 {delay} 秒后重试...")
            time.sleep(delay)
        else:
            logging.error(f"重试 {max_retries} 次后仍然失败: {endpoint}")
            return {"status": "error", "message": str(e)}

return {"status": "error", "message": "Max retries exceeded"}

# 测试
if __name__ == "__main__":
result = call_api_with_retry("/api/users", {"name": "张三"}, max_retries=3)
print(f"最终结果: {result}")

预期输出（每次运行不同）：

最终结果: {'status': 'success', 'data': {'name': '张三'}}

或

最终结果: {'status': 'error', 'message': 'Rate limit exceeded'}

一句话解释：自动重试 3 次，每次失败都记录日志，最后无论成功还是失败都能拿到结果。

项目 3：数据清洗脚本，处理脏数据（15 分钟）

需求：从 CSV 文件读取用户数据，清洗脏数据（空名字、非法年龄），把清洗后的数据写入新文件，出错不影响其他行处理。

# project_3_data_cleaner.py
import logging
import csv
from datetime import datetime

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='data_cleaner.log'
)

class DataCleaningError(Exception):
"""数据清洗错误"""
pass

def validate_user(row, row_num):
"""验证用户数据，返回 (是否有效, 清理后的数据或错误信息)"""
errors = []

# 检查名字
name = row.get('name', '').strip()
if not name:
    errors.append("名字为空")
    name = None

# 检查年龄
try:
    age = int(row.get('age', 0))
    if age < 0 or age > 150:
        errors.append(f"年龄非法: {age}")
        age = None
except ValueError:
    errors.append(f"年龄不是数字: {row.get('age')}")
    age = None

# 检查邮箱格式（简单验证）
email = row.get('email', '').strip()
if '@' not in email:
    errors.append("邮箱格式错误")
    email = None

if errors:
    return False, {"row": row_num, "errors": errors}
return True, {"name": name, "age": age, "email": email}

def clean_user_data(input_file, output_file):
"""清洗用户数据"""
valid_count = 0
invalid_count = 0

try:
    with open(input_file, 'r', encoding='utf-8') as infile:
        reader = csv.DictReader(infile)
        rows = list(reader)
        logging.info(f"读取到 {len(rows)} 条数据")

        valid_users = []

        for idx, row in enumerate(rows, start=1):
            try:
                is_valid, result = validate_user(row, idx)

                if is_valid:
                    valid_users.append(result)
                    valid_count += 1
                    logging.debug(f"第 {idx} 行有效: {result['name']}")
                else:
                    invalid_count += 1
                    logging.warning(f"第 {idx} 行无效: {result['errors']}")

            except Exception as e:
                invalid_count += 1
                logging.error(f"处理第 {idx} 行时发生未知错误: {e}")

        # 写入清洗后的数据
        with open(output_file, 'w', encoding='utf-8', newline='') as outfile:
            fieldnames = ['name', 'age', 'email']
            writer = csv.DictWriter(outfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(valid_users)

        logging.info(f"清洗完成！有效: {valid_count} 条，无效: {invalid_count} 条")
        logging.info(f"清洗后数据已保存到: {output_file}")

        return {"valid": valid_count, "invalid": invalid_count}

except FileNotFoundError:
    logging.error(f"输入文件不存在: {input_file}")
    raise DataCleaningError(f"文件不存在: {input_file}")
except Exception as e:
    logging.critical(f"程序执行出错: {e}", exc_info=True)
    raise

# 创建测试数据
def create_test_data():
"""创建测试用的 CSV 文件"""
test_data = [
    {"name": "张三", "age": "25", "email": "zhangsan@example.com"},
    {"name": "", "age": "30", "email": "test@example.com"},           # 空名字
    {"name": "李四", "age": "-5", "email": "lisi@example.com"},       # 非法年龄
    {"name": "王五", "age": "28", "email": "invalid-email"},          # 错误邮箱
    {"name": "赵六", "age": "35", "email": "zhaoliu@example.com"},
    {"name": "孙七", "age": "abc", "email": "sunqi@example.com"},     # 年龄非数字
]

with open('users_raw.csv', 'w', encoding='utf-8', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=['name', 'age', 'email'])
    writer.writeheader()
    writer.writerows(test_data)
print("测试数据已创建: users_raw.csv")

if __name__ == "__main__":
create_test_data()
result = clean_user_data('users_raw.csv', 'users_clean.csv')
print(f"清洗结果: {result}")

预期输出：

测试数据已创建: users_raw.csv
清洗结果: {'valid': 3, 'invalid': 3}

生成的 users_clean.csv 只包含 3 条有效数据。

一句话解释：一行数据出错不影响其他行，所有问题都被记录日志，最后得到干净的数据。

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：裸 except 语句

# ❌ 错误：捕获所有异常，但不知道是什么错
try:
result = 10 / 0
except:
print("出错了")

# ✅ 正确：指定具体异常类型
try:
result = 10 / 0
except ZeroDivisionError as e:
print(f"除数不能为零: {e}")

原因：裸 except 会捕获所有异常，包括 KeyboardInterrupt（用户按 Ctrl+C），让程序无法正常退出。

坑 2：异常被静默吞掉

# ❌ 错误：捕获异常但什么都不做
try:
result = 10 / 0
except ZeroDivisionError:
pass  # 假装什么都没发生

# ✅ 正确：记录日志或重新抛出
try:
result = 10 / 0
except ZeroDivisionError as e:
logging.error(f"发生错误: {e}")
# 或者重新抛出：raise

原因：静默吞掉异常会让你在排查问题时摸不着头脑。

坑 3：print 代替日志

# ❌ 错误：生产环境用 print
print("用户登录成功")
print("开始处理订单")

# ✅ 正确：使用 logging
logging.info("用户登录成功")
logging.info("开始处理订单")

原因：print 输出到屏幕，程序关了就没了。日志可以输出到文件，保留完整记录。

坑 4：finally 里放可能出错的代码

# ❌ 错误：finally 里也可能抛出异常
try:
file = open('data.txt', 'r')
content = file.read()
finally:
file.close()  # 如果上面 open 失败了，这里会抛出 NameError

# ✅ 正确：把 close 放到 try 里面
try:
with open('data.txt', 'r') as file:  # with 语句自动关闭
    content = file.read()
except FileNotFoundError:
logging.error("文件不存在")

原因：with open() 会自动关闭文件，即使发生异常也不会忘记。

坑 5：日志级别配置错误

# ❌ 错误：生产环境设成 DEBUG，文件会爆掉
logging.basicConfig(level=logging.DEBUG)  # 开发用

# ✅ 正确：生产环境用 INFO 或 WARNING
logging.basicConfig(level=logging.WARNING)
logging.warning("这是一个警告")

性能小贴士：日志性能优化

日志记录会消耗 I/O 资源，高频日志可以用 logging.disable() 临时关闭：

# 性能优化示例
import logging

logging.basicConfig(level=logging.DEBUG)

# 在不需要 debug 日志的模块前关闭
logging.disable(logging.DEBUG)

# 这行不会输出任何内容
logging.debug("这条不会显示")

# 恢复
logging.disable(logging.NOTSET)

调试技巧：pdb 交互式调试

当日志不够用，需要一行一行看时，用 pdb：

import pdb

def buggy_function():
result = 0
for i in range(5):
    result += i
    pdb.set_trace()  # 程序在这里暂停，进入交互式调试
    print(f"i={i}, result={result}")

buggy_function()

运行时会在 pdb.set_trace() 处暂停，输入命令：
- n：执行下一行
- p 变量名：打印变量值
- c：继续执行到下一个断点
- q：退出调试

✏️ 练习题

练习 1（2 分钟）：捕获除零错误

# 题目：修改下面的代码，捕获 ZeroDivisionError 并打印 "除数不能为零"
# 输入：10 / 0
# 预期输出：除数不能为零

def divide(a, b):
# 在这里添加 try/except
return a / b

divide(10, 0)

提示：except 后面跟异常类型，as 后面跟变量名接收错误信息。

练习 2（2 分钟）：添加日志记录

# 题目：在函数执行前后添加日志记录
# 输入：调用 process_data("hello")
# 预期输出：日志显示函数开始和结束

import logging
logging.basicConfig(level=logging.INFO)

def process_data(data):
return data.upper()

提示：logging.info() 可以在任何地方调用。

练习 3（3 分钟）：重写项目 2 的重试逻辑

# 题目：修改下面的函数，让它支持最多 2 次重试

# 输入：调用 call_with_retry(lambda: random.choice([True, False])(), max_retries=2)
# 预期输出：最终返回 True 或 {"error": "All retries failed"}

import random

def call_with_retry(func, max_retries=2):
# 补全这里的逻辑
pass

提示：循环尝试，成功就 return，次数用完再返回错误信息。

练习 4（5 分钟）：串联项目 1 和项目 3

# 题目：用项目 1 的日志方式 + 项目 3 的数据验证方式
# 实现一个"带日志的 CSV 数据验证器"

# 需求：
# 1. 读取 CSV 文件（项目 1 的方式）
# 2. 验证每行数据（项目 3 的方式）
# 3. 记录日志到文件

def validate_csv(input_file):
pass  # 实现这里

提示：两个项目的代码合并就行。

练习 5（5 分钟）：分析报错截图

# 题目：以下报错信息是什么原因？怎么修复？

"""
Traceback (most recent call last):
File "test.py", line 10, in <module>
result = int(user_input)
File "<string>", line 1, in __new__
ValueError: invalid literal for int() with base 10: 'abc'

During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "test.py", line 12, in <module>
except ValueError as e:
NameError: name 'ValueError' is not defined
"""

# 代码是：
user_input = input("请输入数字: ")
try:
result = int(user_input)
except ValueError as e:
print(f"错误: {e}")

提示：检查 except 语句的写法。

作业：做一个「错误追踪日志工具」

需求描述：

写一个工具，用于监控一个文件夹里所有 .txt 文件的读取情况，自动记录：
- 成功读取的文件和内容长度
- 读取失败的文件和失败原因

功能点：

1. 异常处理：文件不存在、权限不足、编码错误等都要捕获
2. 日志记录：所有操作记录到 file_monitor.log
3. 统计报告：最后打印"共处理 X 个文件，成功 Y 个，失败 Z 个"

加分项：
1. 支持指定文件夹路径（而不是硬编码）
2. 统计每个文件的读取耗时

验收标准：
- 能运行 python file_monitor.py ./test_folder
- 生成 file_monitor.log 包含所有操作记录
- 控制台输出统计报告

提交方式：评论区贴代码或 GitHub 链接

📚 总结 + 资源

本章核心

1. 异常处理：try/except/finally 三兄弟让程序优雅地处理错误
2. 自定义异常：用 class MyError(Exception) 定义自己的错误类型
3. 日志系统：logging 模块记录一切，事后能还原案发现场

推荐资源

• 📖 Python 官方文档 - logging：最权威的参考
• 📖 《Python 编程：从入门到实践》第 10 章：错误处理和异常
• 🎬 B 站搜索「Python 异常处理」：大量视频教程可选

互动钩子：

你在项目里有没有遇到过"程序崩了但不知道为啥"的糟心事？后来怎么解决的？评论区聊聊，老粉优先回复！也别忘了转给正在学 Python 的朋友～

下章预告：

这一章我们学会了让程序优雅地出错。下一章我们要讲 Composer 与包管理——你有没有想过，为什么别人能一句话装好一个库，而你只能手动下载？这背后的原理是什么？👇