Python文件操作与异常处理实战指南-平芜编程栈

1. Python文件操作实战指南

文件操作是Python编程中最基础也最重要的技能之一。让我们从最基础的打开文件开始，逐步深入到高级用法。

1.1 文件打开模式详解

Python的open()函数支持多种模式，每种模式都有其特定用途：

# 基本读写模式 file = open('example.txt', 'r') # 只读（默认） file = open('example.txt', 'w') # 写入（会覆盖） file = open('example.txt', 'a') # 追加 file = open('example.txt', 'x') # 独占创建 # 组合模式 file = open('example.txt', 'r+') # 读写（文件必须存在） file = open('example.txt', 'w+') # 读写（会创建或覆盖） file = open('example.txt', 'a+') # 读和追加

重要提示：始终使用with语句处理文件操作，它可以自动管理文件关闭，即使在发生异常时也能确保资源释放。

1.2 文件读取方法对比

Python提供了多种读取文件内容的方法，各有适用场景：

with open('large_file.txt', 'r') as f: # 一次性读取全部内容（小文件适用） content = f.read() # 逐行读取（内存友好） for line in f: process_line(line) # 读取为行列表 lines = f.readlines() # 读取指定字节数 chunk = f.read(1024) # 读取1024字节

实际项目中，处理大文件时应避免使用read()或readlines()，因为它们会一次性加载整个文件到内存。

1.3 高效文件写入技巧

写入文件时，有几个性能优化技巧值得注意：

# 批量写入比多次小写入更高效 lines = ['line1\n', 'line2\n', 'line3\n'] with open('output.txt', 'w') as f: f.writelines(lines) # 比多次调用write()更快 # 需要时手动刷新缓冲区 f.write('important data') f.flush() # 确保数据立即写入磁盘

1.4 二进制文件操作

处理图片、视频等二进制文件时，需要使用'b'模式：

# 复制二进制文件 with open('source.jpg', 'rb') as src, open('copy.jpg', 'wb') as dst: while True: chunk = src.read(4096) # 4KB块读取 if not chunk: break dst.write(chunk)

1.5 文件指针操作

掌握文件指针操作可以实现随机访问：

with open('data.bin', 'rb+') as f: f.seek(10) # 移动到第10字节 print(f.read(5)) # 读取5字节 f.seek(-5, 2) # 从文件末尾前移5字节 f.write(b'END') # 修改最后部分内容

2. Python异常处理深度解析

异常处理是编写健壮程序的关键，Python提供了完善的异常处理机制。

2.1 基础try-except结构

try: risky_operation() except ValueError as e: print(f"值错误: {e}") except (TypeError, IndexError): print("类型或索引错误") except Exception: print("未知错误") else: print("没有异常发生时执行") finally: print("无论是否异常都会执行")

2.2 常见内置异常类型

异常类型	触发场景
ValueError	值不符合预期
TypeError	类型操作错误
IndexError	序列索引越界
KeyError	字典键不存在
FileNotFoundError	文件未找到
ZeroDivisionError	除数为零

2.3 自定义异常实践

创建业务特定的异常能提高代码可读性：

class InvalidTransactionError(Exception): """无效交易异常""" def __init__(self, message, code): super().__init__(message) self.code = code def process_transaction(amount): if amount <= 0: raise InvalidTransactionError("金额必须为正数", 400) try: process_transaction(-100) except InvalidTransactionError as e: print(f"错误代码 {e.code}: {e}")

2.4 异常处理最佳实践

只捕获你能处理的异常
异常信息要具体且有帮助
避免空的except块
使用finally释放资源
考虑异常链（Python 3.3+的raise from语法）

def load_config(): try: with open('config.json') as f: return json.load(f) except FileNotFoundError as e: raise ConfigError("配置文件缺失") from e except json.JSONDecodeError as e: raise ConfigError("配置文件格式错误") from e

3. Python模块导入机制揭秘

Python的模块系统是其强大功能的基础，理解导入机制对项目组织至关重要。

3.1 基础导入方式对比

import math # 基本导入 from math import sqrt # 导入特定对象 from collections import defaultdict as ddict # 别名导入 import numpy as np # 模块别名 # 动态导入 module_name = "json" json = __import__(module_name)

3.2 相对导入与绝对导入

在包内模块中，相对导入是更好的选择：

# 在mypackage/submodule.py中 from . import sibling_module # 同级模块 from .. import parent_module # 父级模块 from .sibling import function # 同级模块中的函数

注意：主模块（name== "main"）不能使用相对导入

3.3 导入路径探索

Python解释器按以下顺序查找模块：

当前目录
PYTHONPATH环境变量指定的目录
Python安装的默认路径

import sys print(sys.path) # 查看当前导入路径 # 临时添加导入路径 sys.path.append('/path/to/your/module')

3.4init.py的现代用法

Python 3.3+中，init.py不再是包的必要条件，但它仍然有重要用途：

# mypackage/__init__.py __all__ = ['module1', 'module2'] # 控制from mypackage import *的行为 # 包级别初始化代码 print("Initializing mypackage") # 提供便捷导入 from .module1 import main_function

3.5 导入钩子与元路径

高级用户可以通过实现导入钩子来自定义导入行为：

class CustomImporter: def find_module(self, fullname, path=None): if fullname == "mylib": return self return None def load_module(self, fullname): # 自定义模块加载逻辑 module = create_module_somehow() sys.modules[fullname] = module return module sys.meta_path.append(CustomImporter())

4. unittest框架全面指南

unittest是Python标准库中的测试框架，借鉴了JUnit的设计理念。

4.1 基本测试用例结构

import unittest class TestStringMethods(unittest.TestCase): @classmethod def setUpClass(cls): """类级别测试夹具，所有测试前执行一次""" cls.shared_resource = create_resource() @classmethod def tearDownClass(cls): """类级别清理""" release_resource(cls.shared_resource) def setUp(self): """每个测试方法前执行""" self.test_str = "hello world" def tearDown(self): """每个测试方法后执行""" del self.test_str def test_upper(self): self.assertEqual(self.test_str.upper(), "HELLO WORLD") def test_isupper(self): self.assertTrue("HELLO".isupper()) self.assertFalse("Hello".isupper()) def test_split(self): self.assertEqual(self.test_str.split(), ['hello', 'world']) with self.assertRaises(TypeError): self.test_str.split(2) if __name__ == '__main__': unittest.main()

4.2 核心断言方法

unittest提供了丰富的断言方法：

方法	检查条件
assertEqual(a, b)	a == b
assertNotEqual(a, b)	a != b
assertTrue(x)	bool(x) is True
assertFalse(x)	bool(x) is False
assertIs(a, b)	a is b
assertIsNot(a, b)	a is not b
assertIsNone(x)	x is None
assertIsNotNone(x)	x is not None
assertIn(a, b)	a in b
assertNotIn(a, b)	a not in b
assertIsInstance(a, b)	isinstance(a, b)
assertNotIsInstance(a, b)	not isinstance(a, b)
assertRaises(exc, callable)	callable引发exc异常

4.3 测试套件组织

对于大型项目，需要组织测试套件：

def suite(): suite = unittest.TestSuite() suite.addTest(TestStringMethods('test_upper')) suite.addTests([ TestStringMethods('test_isupper'), TestStringMethods('test_split') ]) return suite # 或者使用自动发现 loader = unittest.TestLoader() suite = loader.discover('tests', pattern='test_*.py') runner = unittest.TextTestRunner(verbosity=2) runner.run(suite)

4.4 高级测试技巧

跳过测试：

@unittest.skip("暂时跳过此测试") def test_skipped(self): self.fail("不应该执行") @unittest.skipIf(sys.platform == "win32", "不在Windows上运行") def test_not_on_windows(self): pass

子测试：用于参数化测试

def test_even(self): """测试0-5的数字是否为偶数""" for i in range(0, 6): with self.subTest(i=i): self.assertEqual(i % 2, 0)

模拟对象（Python 3.3+的unittest.mock）：

from unittest.mock import Mock, patch class TestPayment(unittest.TestCase): def test_payment_processing(self): payment_gateway = Mock() payment_gateway.process.return_value = True result = process_payment(payment_gateway, 100) self.assertTrue(result) payment_gateway.process.assert_called_once_with(100) @patch('module.payment_gateway') def test_payment_with_patch(self, mock_gateway): mock_gateway.process.return_value = False result = process_payment(mock_gateway, 50) self.assertFalse(result)

4.5 测试覆盖率与持续集成

虽然unittest本身不提供覆盖率统计，但可以结合coverage.py使用：

# 安装coverage pip install coverage # 运行测试并收集覆盖率 coverage run -m unittest discover # 生成报告 coverage report -m coverage html # 生成HTML报告

在CI/CD流程中，典型的配置可能包括：

# .github/workflows/tests.yml 示例 name: Python Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.x' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install coverage - name: Run tests run: | coverage run -m unittest discover coverage report

5. 综合实战：文件操作与异常处理的测试

让我们结合文件操作、异常处理和unittest框架，实现一个完整的测试案例。

5.1 实现文件处理器

# file_processor.py import json class FileProcessor: @staticmethod def read_json_file(filepath): """读取JSON文件并返回解析后的数据""" try: with open(filepath, 'r') as f: return json.load(f) except FileNotFoundError: raise ValueError(f"文件未找到: {filepath}") except json.JSONDecodeError: raise ValueError(f"无效的JSON格式: {filepath}") @staticmethod def write_json_file(filepath, data): """将数据写入JSON文件""" if not isinstance(data, dict): raise TypeError("只支持字典类型数据") try: with open(filepath, 'w') as f: json.dump(data, f, indent=2) return True except IOError as e: raise RuntimeError(f"写入文件失败: {str(e)}")

5.2 编写测试用例

# test_file_processor.py import unittest import tempfile import os from file_processor import FileProcessor class TestFileProcessor(unittest.TestCase): @classmethod def setUpClass(cls): """创建临时测试文件""" cls.temp_dir = tempfile.mkdtemp() cls.valid_json = os.path.join(cls.temp_dir, 'valid.json') cls.invalid_json = os.path.join(cls.temp_dir, 'invalid.json') cls.nonexistent_file = os.path.join(cls.temp_dir, 'nonexistent.json') # 创建有效JSON文件 with open(cls.valid_json, 'w') as f: json.dump({'key': 'value'}, f) # 创建无效JSON文件 with open(cls.invalid_json, 'w') as f: f.write('{"key": "value"') def test_read_valid_json(self): """测试读取有效JSON文件""" data = FileProcessor.read_json_file(self.valid_json) self.assertEqual(data, {'key': 'value'}) def test_read_nonexistent_file(self): """测试读取不存在的文件""" with self.assertRaises(ValueError) as cm: FileProcessor.read_json_file(self.nonexistent_file) self.assertIn("文件未找到", str(cm.exception)) def test_read_invalid_json(self): """测试读取无效JSON文件""" with self.assertRaises(ValueError) as cm: FileProcessor.read_json_file(self.invalid_json) self.assertIn("无效的JSON格式", str(cm.exception)) def test_write_json_success(self): """测试成功写入JSON文件""" test_file = os.path.join(self.temp_dir, 'output.json') test_data = {'test': 'data'} result = FileProcessor.write_json_file(test_file, test_data) self.assertTrue(result) # 验证文件内容 with open(test_file, 'r') as f: content = json.load(f) self.assertEqual(content, test_data) def test_write_invalid_data(self): """测试写入非字典数据""" with self.assertRaises(TypeError): FileProcessor.write_json_file('dummy.json', [1, 2, 3]) @unittest.skipIf(os.name == 'nt', '跳过Windows权限测试') def test_write_permission_error(self): """测试无写入权限的情况""" if os.name == 'posix': read_only_file = os.path.join(self.temp_dir, 'readonly.json') with open(read_only_file, 'w') as f: f.write('{}') os.chmod(read_only_file, 0o444) # 只读权限 with self.assertRaises(RuntimeError): FileProcessor.write_json_file(read_only_file, {'key': 'value'}) @classmethod def tearDownClass(cls): """清理临时文件""" for filename in os.listdir(cls.temp_dir): filepath = os.path.join(cls.temp_dir, filename) try: os.unlink(filepath) except: pass try: os.rmdir(cls.temp_dir) except: pass if __name__ == '__main__': unittest.main(verbosity=2)

5.3 测试覆盖率优化技巧

边界条件测试：文件为空、超大文件、特殊字符等
错误恢复测试：测试程序能否从错误中正确恢复
性能测试：大文件处理性能
并发测试：多线程/多进程环境下的文件操作

# 在TestFileProcessor类中添加 def test_empty_file(self): """测试空文件处理""" empty_file = os.path.join(self.temp_dir, 'empty.json') with open(empty_file, 'w') as f: pass with self.assertRaises(ValueError): FileProcessor.read_json_file(empty_file) def test_large_file(self): """测试大文件处理（不实际创建大文件）""" large_data = {'key': 'x' * 10**6} # 1MB数据 mock_file = 'mock_large.json' with patch('builtins.open', mock_open()) as mock_file: with patch('json.load') as mock_load: mock_load.return_value = large_data data = FileProcessor.read_json_file('dummy.json') self.assertEqual(data, large_data)

6. 高级主题：模块导入与测试的结合

6.1 动态导入测试

对于插件式架构的应用程序，可能需要动态导入并测试模块：

class TestDynamicImports(unittest.TestCase): def test_dynamic_import(self): """测试动态导入的模块""" try: plugin = __import__('my_plugin') self.assertTrue(hasattr(plugin, 'main_function')) except ImportError: self.skipTest("插件模块不可用") def test_import_error_handling(self): """测试导入错误处理""" with self.assertRaises(ImportError): __import__('nonexistent_module')

6.2 模拟导入行为

在测试中模拟导入行为可以隔离测试环境：

class TestImportMocking(unittest.TestCase): @patch.dict('sys.modules', {'external_lib': None}) def test_without_external_lib(self): """模拟缺少外部依赖的情况""" with self.assertRaises(ImportError): from external_lib import important_function important_function() @patch('module.imported_function') def test_mock_imported_function(self, mock_func): """模拟导入的函数""" mock_func.return_value = 42 from module import do_something result = do_something() self.assertEqual(result, 42)

6.3 测试导入性能

对于大型项目，导入时间可能成为问题，可以添加导入性能测试：

class TestImportPerformance(unittest.TestCase): def test_import_time(self): """测试关键模块的导入时间""" import time start = time.perf_counter() import numpy # 示例：测试numpy导入时间 elapsed = time.perf_counter() - start self.assertLess(elapsed, 1.0, "导入时间过长")

7. 常见问题与解决方案

7.1 文件操作常见错误

编码问题：

# 指定编码避免问题 with open('file.txt', 'r', encoding='utf-8') as f: content = f.read()

资源泄漏：

# 错误示范 f = open('file.txt') # 可能泄漏 # 正确做法 with open('file.txt') as f: pass

路径问题：

# 使用os.path处理路径 import os file_path = os.path.join('dir', 'subdir', 'file.txt')

7.2 异常处理陷阱

过于宽泛的异常捕获：

# 错误示范 try: do_something() except: # 捕获所有异常，包括SystemExit pass # 正确做法 try: do_something() except (ValueError, TypeError) as e: handle_error(e)

忽略异常：

# 错误示范 try: do_something() except Error: pass # 静默忽略 # 更好做法 try: do_something() except Error as e: log_error(e) raise # 重新抛出或处理

7.3 unittest常见问题

测试顺序依赖：
- 每个测试方法应该是独立的
- 使用setUp()确保干净的测试环境
缓慢的测试：
- 使用mock替换慢速操作
- 将单元测试与集成测试分开

测试失败信息不足：

# 不够好 self.assertEqual(result, expected) # 更好 self.assertEqual(result, expected, f"对于输入{input_data}，期望{expected}但得到{result}")

7.4 模块导入问题

循环导入：
- 重构代码消除循环依赖
- 将导入移到函数内部（延迟导入）

Python路径问题：

# 调试导入问题 import sys print(sys.path) # 临时添加路径 sys.path.insert(0, '/path/to/your/module')

相对导入问题：
- 在包内使用相对导入
- 主模块使用绝对导入

8. 性能优化与最佳实践

8.1 文件操作性能优化

缓冲策略：

# 调整缓冲区大小（默认通常是8KB） with open('large.bin', 'rb', buffering=64*1024) as f: # 64KB缓冲区 data = f.read()

内存映射文件：

import mmap with open('large.bin', 'r+b') as f: mm = mmap.mmap(f.fileno(), 0) # 像操作内存一样访问文件 print(mm[10:20]) mm.close()

批量操作：

# 批量写入比单次写入高效 lines = [f"line{i}\n" for i in range(10000)] with open('big.txt', 'w') as f: f.writelines(lines) # 比多次write()快

8.2 异常处理性能

异常 vs 条件检查：

# 在频繁执行的代码中，条件检查可能比捕获异常更高效 if key in my_dict: # 比try-except更快 value = my_dict[key]

避免深层嵌套：

# 难以维护的深层嵌套 try: try: try: ... except Error1: ... except Error2: ... except Error3: ... # 更清晰的结构 def step1(): try: ... except Error1: ... def step2(): try: step1() except Error2: ... try: step2() except Error3: ...

8.3 测试套件优化

测试分类：

# 创建不同的测试套件 fast_suite = unittest.TestSuite() slow_suite = unittest.TestSuite() # 根据测试速度分类 for test in all_tests: if is_slow_test(test): slow_suite.addTest(test) else: fast_suite.addTest(test)

并行测试：

# 使用concurrent.futures并行运行测试 import concurrent.futures def run_test(test): runner = unittest.TextTestRunner(stream=open('/dev/null', 'w')) return runner.run(test) with concurrent.futures.ProcessPoolExecutor() as executor: results = list(executor.map(run_test, test_suites))

测试数据管理：

# 使用setUpModule和tearDownModule def setUpModule(): global test_data test_data = generate_large_dataset() def tearDownModule(): global test_data del test_data

9. 现代Python测试工具链

虽然unittest是标准库，但现代Python项目通常会结合其他工具：

9.1 pytest集成

pytest可以与unittest测试共存并提供更多功能：

# 安装pytest pip install pytest # 运行unittest测试（兼容） pytest tests/ # pytest特性示例 def test_with_pytest(): assert 1 + 1 == 2 # 参数化测试 import pytest @pytest.mark.parametrize("input,expected", [ ("3+5", 8), ("2+4", 6), ("6*9", 42), ]) def test_eval(input, expected): assert eval(input) == expected

9.2 测试覆盖率工具

# 安装pytest-cov pip install pytest-cov # 运行测试并收集覆盖率 pytest --cov=myproject tests/ # 生成HTML报告 pytest --cov=myproject --cov-report=html tests/

9.3 基准测试

使用pytest-benchmark进行性能测试：

# conftest.py import pytest from myproject import process_data @pytest.fixture def large_dataset(): return [i for i in range(10**6)] def test_process_performance(benchmark, large_dataset): result = benchmark(process_data, large_dataset) assert result is not None

9.4 类型检查测试

结合mypy进行静态类型检查：

# 安装mypy pip install mypy # 运行类型检查 mypy myproject/ # 在测试中验证类型 from typing import Any def test_type_annotations(): from myproject import some_function assert some_function.__annotations__ == {'param': int, 'return': str}

10. 项目结构建议

合理的项目结构有助于管理测试和模块：

myproject/ ├── src/ │ ├── mypackage/ │ │ ├── __init__.py │ │ ├── module1.py │ │ └── module2.py ├── tests/ │ ├── __init__.py │ ├── unit/ │ │ ├── test_module1.py │ │ └── test_module2.py │ └── integration/ │ ├── test_api.py │ └── test_db.py ├── pyproject.toml └── README.md

关键点：

将测试与源代码分离
区分单元测试和集成测试
每个测试文件对应一个源文件
测试模块名以test_开头
测试类名以Test开头
测试方法名以test_开头

11. 持续集成配置示例

GitHub Actions的Python测试工作流示例：

name: Python Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest strategy: matrix: python-version: ["3.8", "3.9", "3.10"] steps: - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | python -m pip install --upgrade pip pip install -e .[test] - name: Run tests with pytest run: | pytest --cov=./ --cov-report=xml - name: Upload coverage uses: codecov/codecov-action@v3 - name: Run mypy run: | mypy src/

12. 调试技巧

12.1 测试调试

pdb调试：

def test_debug_example(self): import pdb; pdb.set_trace() # 设置断点 result = complex_operation() self.assertEqual(result, expected)

失败重运行：

pytest --lf # 只运行上次失败的测试 pytest --ff # 先运行上次失败的测试

12.2 文件操作调试

# 检查文件状态 import os print(os.stat('file.txt')) print(os.access('file.txt', os.R_OK)) # 检查读权限

12.3 导入调试

# 查看模块加载过程 import importlib.util import sys def debug_import(name): print(f"尝试导入 {name}") spec = importlib.util.find_spec(name) print(f"找到的spec: {spec}") if spec: print(f"加载器: {spec.loader}") print(f"源文件: {spec.origin}") module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module return None debug_import('mymodule')

13. 安全注意事项

13.1 文件操作安全

路径遍历防护：

from pathlib import Path def secure_open(base_dir, filename): path = (Path(base_dir) / filename).resolve() if not path.is_relative_to(Path(base_dir).resolve()): raise ValueError("非法路径访问") return open(path)

临时文件安全：

import tempfile # 安全创建临时文件 with tempfile.NamedTemporaryFile(delete=True) as tmp: tmp.write(b'data') tmp.flush() # 使用临时文件

13.2 测试中的安全

隔离测试环境：
- 使用虚拟环境
- 不要在生产环境中运行测试
- 测试数据库使用专用实例

敏感数据处理：

# 不要在测试中硬编码真实凭证 @patch.dict('os.environ', {'DB_PASSWORD': 'test_password'}) def test_database_connection(self): connect_to_db()

14. 跨平台考虑

14.1 文件路径处理

from pathlib import Path # 跨平台路径构造 config_path = Path('config') / 'settings.ini' # 路径比较 if some_path == Path('/expected/path'): pass

14.2 行尾符处理

# 统一行尾符 with open('file.txt', 'r', newline='') as f: content = f.read() # 不转换行尾符

14.3 编码问题

# 显式指定编码 with open('file.txt', 'r', encoding='utf-8') as f: content = f.read()

15. 性能测试实战

15.1 文件IO性能测试

import unittest import tempfile import os import timeit class TestFilePerformance(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp_file = tempfile.NamedTemporaryFile(delete=False) cls.temp_file.close() # 准备1MB测试数据 cls.test_data = b'x' * 1024 * 1024 def test_write_performance(self): def write_test(): with open(self.temp_file.name, 'wb') as f: f.write(self.test_data) time = timeit.timeit(write_test, number=100) self.assertLess(time, 1.0, "写入性能不达标") def test_read_performance(self): # 先写入测试数据 with open(self.temp_file.name, 'wb') as f: f.write(self.test_data) def read_test(): with open(self.temp_file.name, 'rb') as f: data = f.read() time = timeit.timeit(read_test, number=100) self.assertLess(time, 1.0, "读取性能不达标") @classmethod def tearDownClass(cls): try: os.unlink(cls.temp_file.name) except: pass

15.2 导入时间测试

class TestImportPerformance(unittest.TestCase): def test_import_time(self): import time modules_to_test = ['json', 'csv', 're'] for module in modules_to_test: with self.subTest(module=module): start = time.perf_counter() __import__(module) elapsed = time.perf_counter() - start self.assertLess(elapsed, 0.1, f"导入 {module} 耗时 {elapsed:.3f} 秒")

16. 资源管理进阶

16.1 上下文管理器实现

class DatabaseConnection: def __init__(self, connection_string): self.connection_string = connection_string self.connection = None def __enter__(self): self.connection = connect_to_db(self.connection_string) return self.connection def __exit__(self, exc_type, exc_val, exc_tb): if self.connection: self.connection.close() if exc_type is not None: print(f"发生错误: {exc_val}") return False # 不抑制异常 # 测试上下文管理器 class TestDatabaseConnection(unittest.TestCase): def test_context_manager(self): with DatabaseConnection('test://localhost') as conn: self.assertIsNotNone(conn) self.assertTrue(conn.is_connected()) # 测试连接是否已关闭 self.assertFalse(conn.is_connected()) def test_exception_handling(self): with self.assertRaises(DatabaseError): with DatabaseConnection('invalid://') as conn: raise DatabaseError("连接失败")

16.2 使用atexit进行清理

import atexit import tempfile class TempFileManager: _files_to_clean = set() @classmethod def create_temp(cls): tmp = tempfile.NamedTemporaryFile(delete=False) cls._files_to_clean.add(tmp.name) return tmp.name @