06｜持久化存储与防御性编程：文件IO与异常机制

场景引入：数据导出服务的健壮性需求

你开发了一个数据分析脚本，跑了两个小时终于算完了结果——但程序在写入文件时崩溃了，所有数据丢失。更糟糕的是，崩溃原因只是用户传入了一个不存在的目录路径。

这个教训引出两个工程课题：

1. 文件操作：如何正确地读写数据，让信息持久化到磁盘
2. 异常处理：如何让程序在遇到意外状况时不崩溃，而是给出有意义的反馈

---

文件读写基础

open() 函数

Python 用 open() 函数打开文件：

handle = open('路径', '模式')

常用模式：

模式	含义	说明
'r'	读取（默认）	文件必须已存在
'w'	写入	不存在则创建，已存在则清空
'a'	追加	不存在则创建，已存在则在末尾添加
'rb'	二进制读取	用于图片、音频等非文本文件
'wb'	二进制写入	写入二进制内容

读取文件

handle = open('report.txt', 'r')
content = handle.read()
print(content)
handle.close()        # 使用完毕必须关闭

**为什么必须关闭？**操作系统对同时打开的文件数量有上限。未关闭的文件句柄会占用系统资源，可能导致数据未完整写入磁盘。

不同读取方式适用于不同场景：

handle = open('log.txt', 'r', encoding='utf-8')

# read() —— 全量读取（适合小文件）
full_text = handle.read()

# read(size) —— 按字节数读取（适合大文件分块处理）
chunk = handle.read(1024)

# readline() —— 读取单行
single_line = handle.readline()

# readlines() —— 读取全部行，返回列表
all_lines = handle.readlines()

handle.close()

with 语句：自动资源管理

手动调用 close() 容易遗忘，且在异常发生时可能被跳过。with 语句自动管理文件的打开和关闭：

with open('report.txt', 'r', encoding='utf-8') as handle:
    content = handle.read()
    print(content)
# 离开 with 块后，文件自动关闭

# 逐行处理大文件的最佳方式
with open('access.log', 'r', encoding='utf-8') as handle:
    for line in handle:
        print(line.strip())

写入文件

# 写入模式：覆盖原有内容
with open('result.txt', 'w', encoding='utf-8') as handle:
    handle.write('第一行数据\n')
    handle.write('第二行数据\n')

# 追加模式：在末尾添加
with open('result.txt', 'a', encoding='utf-8') as handle:
    handle.write('追加的第三行\n')

# 批量写入多行
records = ['alpha\n', 'beta\n', 'gamma\n']
with open('records.txt', 'w', encoding='utf-8') as handle:
    handle.writelines(records)    # writelines 不自动添加换行符

write() 不会自动添加换行符，需要手动写入 \n。

字符编码

不同系统和文件可能使用不同的编码方式。编码不匹配会导致乱码或报错：

# UTF-8 编码（最通用）
with open('data.txt', 'r', encoding='utf-8') as handle:
    content = handle.read()

# GBK 编码（常见于旧版 Windows 中文文件）
with open('legacy.txt', 'r', encoding='gbk') as handle:
    content = handle.read()

# 容忍少量编码错误
with open('mixed.txt', 'r', encoding='utf-8', errors='ignore') as handle:
    content = handle.read()

---

处理常见数据格式

CSV 文件

CSV（逗号分隔值）是数据交换中最常见的纯文本格式：

import csv

# 读取
with open('inventory.csv', 'r', encoding='utf-8') as handle:
    reader = csv.reader(handle)
    header = next(reader)
    print(f'字段：{header}')
    for row in reader:
        print(f'品名：{row[0]}，数量：{row[1]}')

# 用 DictReader 将每行解析为字典
with open('inventory.csv', 'r', encoding='utf-8') as handle:
    reader = csv.DictReader(handle)
    for row in reader:
        print(f"品名：{row['item']}，数量：{row['quantity']}")

# 写入
with open('export.csv', 'w', encoding='utf-8', newline='') as handle:
    writer = csv.writer(handle)
    writer.writerow(['item', 'quantity'])
    writer.writerow(['keyboard', 45])
    writer.writerow(['mouse', 120])

JSON 文件

JSON 是 Web API 和配置文件中的主流格式：

import json

# 读取
with open('settings.json', 'r', encoding='utf-8') as handle:
    data = json.load(handle)
    print(data['app_name'])

# 写入
profile = {'emp_id': 'E042', 'dept': '研发', 'skills': ['Python', 'SQL']}
with open('profile.json', 'w', encoding='utf-8') as handle:
    json.dump(profile, handle, ensure_ascii=False, indent=2)

# 字符串与对象互转
json_text = '{"status": "active", "level": 3}'
obj = json.loads(json_text)
back_to_text = json.dumps(obj, ensure_ascii=False)

---

文件路径管理

os.path 模块（传统方式）

import os

# 路径拼接
full_path = os.path.join('data', 'exports', 'report.csv')

# 获取文件名和目录
filepath = '/srv/app/data/report.csv'
print(os.path.basename(filepath))   # report.csv
print(os.path.dirname(filepath))    # /srv/app/data

# 分离文件名和扩展名
stem, ext = os.path.splitext('snapshot.tar.gz')

# 判断路径是否存在
print(os.path.exists('/tmp/test.txt'))
print(os.path.isfile('/tmp/test.txt'))
print(os.path.isdir('/tmp'))

# 当前工作目录
print(os.getcwd())

# 列出目录内容
print(os.listdir('.'))

pathlib 模块（推荐方式）

pathlib 以面向对象的风格操作路径，代码更清晰：

from pathlib import Path

# 路径拼接
target = Path('data') / 'exports' / 'report.csv'
print(target)

# 获取路径各部分
fp = Path('/srv/app/data/report.csv')
print(fp.name)       # report.csv
print(fp.stem)       # report
print(fp.suffix)     # .csv
print(fp.parent)     # /srv/app/data

# 路径判断
print(fp.exists())
print(fp.is_file())
print(fp.is_dir())

# 快速读写
p = Path('memo.txt')
p.write_text('备忘内容', encoding='utf-8')
content = p.read_text(encoding='utf-8')

# 特殊路径
print(Path.cwd())     # 当前工作目录
print(Path.home())    # 用户主目录

# 文件搜索
for csv_file in Path('.').glob('*.csv'):
    print(csv_file)

for py_file in Path('.').glob('**/*.py'):
    print(py_file)

# 创建目录
Path('output/charts').mkdir(parents=True, exist_ok=True)

---

异常处理机制

什么是异常？

程序运行时遇到无法继续的错误就会抛出异常。不处理的异常会导致程序直接终止：

print(100 / 0)            # ZeroDivisionError
print(int('xyz'))          # ValueError
print(undefined_var)       # NameError
print([1, 2][10])          # IndexError
print({'a': 1}['b'])       # KeyError

try/except 捕获异常

try:
    divisor = int(input('输入除数：'))
    quotient = 1000 / divisor
    print(f'结果：{quotient}')
except ValueError:
    print('请输入有效的整数')
except ZeroDivisionError:
    print('除数不能为零')

执行流程：

1. 执行 try 块
2. 若无异常，跳过所有 except
3. 若有异常，跳转到匹配的 except 块
4. 若没有匹配的 except，异常继续向上传播

捕获多种异常和获取详情

# 同时捕获多种异常
try:
    handle = open('data.bin', 'r')
    val = int(handle.read())
except (FileNotFoundError, ValueError) as err:
    print(f'处理失败：{err}')

# 捕获所有异常（调试时使用，生产环境慎用）
try:
    risky_operation()
except Exception as err:
    print(f'{type(err).__name__}: {err}')

try/except/else/finally 完整结构

def safe_division(numerator, denominator):
    try:
        result = numerator / denominator
    except ZeroDivisionError:
        print('错误：除数为零')
    else:
        # 仅在 try 块无异常时执行
        print(f'结果：{result}')
    finally:
        # 无论是否异常都执行
        print('运算结束')

safe_division(100, 3)
# 结果：33.333333333333336
# 运算结束

safe_division(100, 0)
# 错误：除数为零
# 运算结束

finally 的典型用途是释放资源（关闭文件、断开连接等），确保清理逻辑一定执行。

常见异常类型

异常	说明	触发示例
ValueError	值不合法	int('xyz')
TypeError	类型不匹配	'5' + 5
KeyError	字典键不存在	d['missing']
IndexError	索引越界	[1,2][10]
FileNotFoundError	文件不存在	open('ghost.txt')
ZeroDivisionError	除以零	1 / 0
AttributeError	属性不存在	"text".nonexist()
ImportError	导入失败	import ghost_module
NameError	变量未定义	print(undefined)

raise：主动抛出异常

def set_threshold(val):
    if not isinstance(val, (int, float)):
        raise TypeError('阈值必须是数字类型')
    if val < 0 or val > 100:
        raise ValueError('阈值必须在 0-100 之间')
    print(f'阈值已设置为：{val}')

try:
    set_threshold(-10)
except ValueError as err:
    print(f'设置失败：{err}')

自定义异常类

class QuotaExceededError(Exception):
    """配额超限异常"""
    def __init__(self, current, limit):
        self.current = current
        self.limit = limit
        super().__init__(f'配额已满：当前 {current}，上限 {limit}')

class ResourcePool:
    def __init__(self, capacity):
        self.capacity = capacity
        self.used = 0

    def allocate(self, amount):
        if self.used + amount > self.capacity:
            raise QuotaExceededError(self.used, self.capacity)
        self.used += amount
        print(f'分配成功，已使用：{self.used}/{self.capacity}')

pool = ResourcePool(100)
try:
    pool.allocate(80)
    pool.allocate(30)
except QuotaExceededError as err:
    print(err)

---

logging 模块：专业的日志记录

生产环境中，用 print() 做调试既不灵活也不可控。logging 模块提供了分级日志机制：

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

logging.debug('变量值：payload_size = 1024')    # 开发调试
logging.info('服务启动成功')                      # 运行状态
logging.warning('磁盘使用率超过 80%')            # 潜在问题
logging.error('数据库连接失败')                   # 运行错误
logging.critical('内存耗尽，服务终止')            # 致命错误

日志级别：DEBUG < INFO < WARNING < ERROR < CRITICAL。设置 level=logging.WARNING 后，DEBUG 和 INFO 级别的日志不会输出。

# 输出到文件
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='service.log',
    filemode='a'
)

# 在异常处理中使用
def load_config(filepath):
    try:
        with open(filepath, 'r', encoding='utf-8') as handle:
            data = handle.read()
        logging.info(f'配置加载成功：{filepath}')
        return data
    except FileNotFoundError:
        logging.error(f'配置文件不存在：{filepath}')
        return None
    except Exception as err:
        logging.exception(f'加载配置时异常：{filepath}')
        return None

---

综合实战：健壮的 JSON 文件读取器

import json
import logging
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def load_json_safe(filepath):
    """健壮的 JSON 文件读取函数"""
    target = Path(filepath)

    if not target.exists():
        logging.error(f'文件不存在：{filepath}')
        return None

    if target.suffix.lower() != '.json':
        logging.warning(f'文件扩展名非 .json：{filepath}')

    try:
        with open(target, 'r', encoding='utf-8') as handle:
            data = json.load(handle)
        logging.info(f'JSON 文件读取成功：{filepath}')
        return data
    except json.JSONDecodeError as err:
        logging.error(f'JSON 格式错误：{err}')
        return None
    except UnicodeDecodeError:
        logging.error(f'编码错误，请确认文件为 UTF-8 编码')
        return None

# 使用
config_data = load_json_safe('app_config.json')
if config_data:
    print(f"应用名称：{config_data.get('app_name', '未指定')}")

---

本章回顾

文件操作核心：

• 使用 with open(...) as f 确保资源自动释放
• 始终指定 encoding='utf-8' 防止编码问题
• 用 pathlib.Path 替代 os.path 处理路径
• 掌握 CSV 和 JSON 两种主流数据格式的读写

异常处理核心：

• try/except 捕获异常，防止程序崩溃
• finally 确保清理代码一定执行
• else 在无异常时执行额外逻辑
• 精确捕获特定异常类型，避免裸 except
• raise 主动抛出异常，自定义异常类表达业务错误
• 用 logging 替代 print 进行日志记录

补充：异常处理的最佳实践

1. 精确捕获：始终捕获具体的异常类型，而非使用裸的 except:

# 不推荐
try:
    process()
except:
    pass

# 推荐
try:
    process()
except (ValueError, KeyError) as err:
    logging.error(f"处理失败：{err}")

2. 不要用异常做流程控制：异常应该用于处理”意外情况”，而非替代 if/else 做常规的逻辑分支

3. 尽早失败：如果输入数据不合法，在函数入口处就抛出异常，而不是等到中间环节才出错

def process_batch(items):
    if not isinstance(items, list):
        raise TypeError("items 必须是列表类型")
    if len(items) == 0:
        raise ValueError("items 不能为空")
    # 正式处理逻辑...

4. 清理资源：涉及外部资源（文件、网络连接、数据库游标）的代码，务必使用 with 语句或 finally 确保清理

补充：文件操作的常见模式

from pathlib import Path

# 读取文件并按行处理，跳过空行和注释行
def parse_config_lines(filepath):
    result = []
    with open(filepath, 'r', encoding='utf-8') as handle:
        for line in handle:
            stripped = line.strip()
            if not stripped or stripped.startswith('#'):
                continue
            result.append(stripped)
    return result

# 安全写入：先写到临时文件，再重命名（防止写入中途崩溃导致数据丢失）
import tempfile
import shutil

def safe_write(filepath, content):
    target = Path(filepath)
    with tempfile.NamedTemporaryFile('w', dir=target.parent,
                                     delete=False, encoding='utf-8') as tmp:
        tmp.write(content)
        tmp_path = tmp.name
    shutil.move(tmp_path, target)

核心理念：优秀的程序不是从不出错的程序，而是能在异常情况下做出合理响应的程序。

下一章将学习模块和包的组织方式——如何复用已有代码、管理项目依赖、利用 Python 丰富的标准库加速开发。