04-上下文管理器
Python 版本要求： 本教程基于 Python 3.11+ 编写，部分特性需要 3.11 或更高版本。
导航： 错误与异常基础 → 异常处理 → 抛出异常 → 上下文管理器
概念铺垫
第一步：实际场景引入
为什么需要上下文管理器？
假设你需要处理文件操作：
python
def read_file_bad(filepath: str) -> str:
"""不安全的方式"""
file = open(filepath, "r", encoding="utf-8")
content = file.read()
# 如果这里发生异常，文件永远不会被关闭！
file.close()
return content问题来了：
场景 1: 文件读取时发生异常
→ file.close() 不会执行
→ 文件句柄泄漏 💥
场景 2: 需要处理多个资源
→ 数据库连接、锁、网络连接
→ 手动管理非常繁琐
场景 3: 异常处理嵌套
→ try-finally 嵌套多层
→ 代码难以阅读传统解决方案：
python
def read_file_traditional(filepath: str) -> str | None:
"""传统方式：繁琐且容易出错"""
file = None
try:
file = open(filepath, "r", encoding="utf-8")
content = file.read()
return content
except FileNotFoundError:
print(f"文件不存在：{filepath}")
return None
finally:
if file is not None:
file.close()思考： 如何简化资源管理，确保资源总能被正确释放？
第二步：概念动机
上下文管理器的核心思想
┌─────────────────────────────────────────┐
│ 上下文管理器的核心思想 │
├─────────────────────────────────────────┤
│ │
│ 核心概念： │
│ ───────────────────────── │
│ 定义一个上下文，在进入和退出时 │
│ 自动执行特定操作。 │
│ │
│ ───────────────────────────────── │
│ │
│ with 语句执行流程： │
│ │
│ 1. 进入上下文 │
│ • 调用 __enter__ │
│ • 获取资源 │
│ │
│ 2. 执行代码块 │
│ • 正常执行或发生异常 │
│ │
│ 3. 退出上下文 │
│ • 调用 __exit__ │
│ • 释放资源 │
│ • 处理异常（如果需要） │
│ │
│ ───────────────────────────────── │
│ │
│ 典型应用场景： │
│ • 文件操作 │
│ • 数据库连接 │
│ • 锁的获取和释放 │
│ • 计时和性能测量 │
│ • 临时状态修改 │
│ │
└─────────────────────────────────────────┘核心理解：
- with：创建上下文
- enter：进入时执行
- exit：退出时执行（总是执行）
L1 理解层：会用
第三步：最简示例
使用 with 语句
python
# 文件操作
with open("file.txt", "r", encoding="utf-8") as f:
content: str = f.read()
# 文件自动关闭，即使发生异常
# 对比传统方式
file = None
try:
file = open("file.txt", "r", encoding="utf-8")
content = file.read()
finally:
if file:
file.close()第四步：详细讲解
with 语句
基本用法
python
# 文件操作
def read_file_content(filepath: str) -> str | None:
"""安全读取文件"""
try:
with open(filepath, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
print(f"文件不存在：{filepath}")
return None
# 写入文件
def write_file_content(filepath: str, content: str) -> bool:
"""安全写入文件"""
try:
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
return True
except IOError as e:
print(f"写入失败：{e}")
return False多个上下文管理器
python
from pathlib import Path
def copy_file_safe(src: str, dst: str) -> bool:
"""安全复制文件"""
try:
with open(src, "r", encoding="utf-8") as fin, \
open(dst, "w", encoding="utf-8") as fout:
fout.write(fin.read())
return True
except FileNotFoundError:
print(f"源文件不存在：{src}")
return False
except IOError as e:
print(f"复制失败：{e}")
return False
# Python 3.10+ 多行写法
def copy_file_modern(src: str, dst: str) -> bool:
"""Python 3.10+ 多行写法"""
try:
with (
open(src, "r", encoding="utf-8") as fin,
open(dst, "w", encoding="utf-8") as fout,
):
fout.write(fin.read())
return True
except Exception as e:
print(f"复制失败：{e}")
return False上下文管理器协议
概念说明
上下文管理器需要实现两个方法：
__enter__：进入上下文时调用__exit__：离开上下文时调用
python
from typing import Optional, Any
class ManagedFile:
"""文件上下文管理器"""
def __init__(self, filename: str, mode: str) -> None:
self.filename: str = filename
self.mode: str = mode
self.file: Optional[Any] = None
def __enter__(self) -> Any:
"""进入上下文时调用"""
print(f"打开文件：{self.filename}")
self.file = open(self.filename, self.mode, encoding="utf-8")
return self.file
def __exit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[Any]
) -> bool:
"""离开上下文时调用"""
print("关闭文件...")
if self.file:
self.file.close()
return False # 不吞掉异常
# 使用
def read_with_managed(filepath: str) -> str | None:
"""使用自定义上下文管理器"""
try:
with ManagedFile(filepath, "r") as f:
return f.read()
except FileNotFoundError:
print(f"文件不存在：{filepath}")
return Noneexit 参数说明
python
from typing import Optional, Any
class DebugContext:
"""调试上下文管理器"""
def __init__(self, name: str) -> None:
self.name: str = name
def __enter__(self) -> "DebugContext":
print(f"进入 {self.name}")
return self
def __exit__(
self,
exc_type: Optional[type], # 异常类型
exc_val: Optional[BaseException], # 异常值
exc_tb: Optional[Any] # 异常追踪信息
) -> bool:
if exc_type:
print(f"{self.name} 发生异常：{exc_type.__name__}: {exc_val}")
else:
print(f"正常退出 {self.name}")
# 返回 True 表示异常已处理，不会继续传播
# 返回 False 表示异常继续传播
return False
# 使用
def example_debug() -> None:
with DebugContext("测试区块"):
print("执行代码...")
# 如果这里发生异常，exc_type 不为 None
# 捕获并处理异常
class SuppressError:
"""抑制特定异常的上下文管理器"""
def __init__(self, *exceptions: type) -> None:
self.exceptions: tuple[type, ...] = exceptions
def __enter__(self) -> "SuppressError":
return self
def __exit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[Any]
) -> bool:
if exc_type and issubclass(exc_type, self.exceptions):
print(f"抑制异常：{exc_type.__name__}: {exc_val}")
return True # 异常已处理
return False # 异常继续传播
# 使用
def example_suppress() -> None:
with SuppressError(ZeroDivisionError, FileNotFoundError):
result = 10 / 0 # 异常被抑制
print("这行不会执行")
print("继续执行...")contextlib 模块
@contextmanager 装饰器
python
from contextlib import contextmanager
from typing import Generator, Any
@contextmanager
def managed_file(filename: str, mode: str) -> Generator[Any, None, None]:
"""文件上下文管理器（生成器版本）"""
f = None
try:
print(f"打开文件：{filename}")
f = open(filename, mode, encoding="utf-8")
yield f # 将文件对象传递给 with 块
finally:
if f:
print("关闭文件...")
f.close()
# 使用
def read_with_context_manager(filepath: str) -> str | None:
"""使用装饰器创建的上下文管理器"""
try:
with managed_file(filepath, "r") as f:
return f.read()
except FileNotFoundError:
print(f"文件不存在：{filepath}")
return Nonecontextlib 常用工具
python
from contextlib import contextmanager, suppress, redirect_stdout
from typing import Generator
import io
import os
# suppress: 抑制异常
def example_suppress() -> None:
# 方式 1：使用 contextlib.suppress
with suppress(FileNotFoundError):
os.remove("不存在的文件.txt")
# 不会抛出异常
# 方式 2：等价于
try:
os.remove("不存在的文件.txt")
except FileNotFoundError:
pass
# redirect_stdout: 重定向标准输出
def example_redirect() -> None:
output: io.StringIO = io.StringIO()
with redirect_stdout(output):
print("这行输出被捕获")
print("这行也被捕获")
captured: str = output.getvalue()
print(f"捕获的内容：{captured}")
# nested: 嵌套上下文管理器
@contextmanager
def database_connection(host: str) -> Generator[dict[str, str], None, None]:
"""模拟数据库连接"""
print(f"连接数据库：{host}")
yield {"host": host, "connected": True}
print("断开数据库连接")
@contextmanager
def transaction(conn: dict[str, str]) -> Generator[dict[str, str], None, None]:
"""模拟事务"""
print("开始事务")
try:
yield conn
print("提交事务")
except Exception:
print("回滚事务")
raise
# 使用嵌套上下文
def example_nested() -> None:
with database_connection("localhost") as conn:
with transaction(conn):
print(f"执行操作：{conn}")第五步：渐进复杂化
构建可复用的上下文管理器
python
import time
from contextlib import contextmanager
from typing import Generator, Any, Callable
# 计时器
@contextmanager
def timer(name: str = "操作") -> Generator[None, None, None]:
"""计时代码执行时间"""
start: float = time.time()
try:
yield
finally:
end: float = time.time()
print(f"{name}耗时：{end - start:.4f} 秒")
# 使用
def example_timer() -> None:
with timer("数据处理"):
result: list[int] = [x * 2 for x in range(1000000)]
# 输出：数据处理耗时：0.xxxx 秒
# 临时切换目录
import os
from pathlib import Path
@contextmanager
def working_directory(path: str | Path) -> Generator[Path, None, None]:
"""临时切换工作目录"""
original_dir: str = os.getcwd()
target_path: Path = Path(path)
try:
os.chdir(target_path)
yield target_path
finally:
os.chdir(original_dir)
# 使用
def example_directory() -> None:
print(f"当前目录：{os.getcwd()}")
with working_directory("/tmp"):
print(f"临时目录：{os.getcwd()}")
print(f"恢复目录：{os.getcwd()}")
# 锁管理
import threading
@contextmanager
def acquire_lock(lock: threading.Lock, timeout: float = -1) -> Generator[bool, None, None]:
"""获取锁的上下文管理器"""
acquired: bool = lock.acquire(timeout=timeout if timeout > 0 else None)
try:
yield acquired
finally:
if acquired:
lock.release()
# 使用
def example_lock() -> None:
lock: threading.Lock = threading.Lock()
with acquire_lock(lock) as acquired:
if acquired:
print("执行临界区代码")
else:
print("获取锁超时")
# 性能分析
@contextmanager
def profile(
name: str = "代码块",
print_result: bool = True
) -> Generator[list[str], None, None]:
"""性能分析上下文管理器"""
logs: list[str] = []
start: float = time.time()
# 记录开始信息
logs.append(f"开始执行：{name}")
try:
yield logs
finally:
end: float = time.time()
duration: float = end - start
# 记录结束信息
logs.append(f"执行完成：{name}")
logs.append(f"耗时：{duration:.4f} 秒")
if print_result:
for log in logs:
print(log)
# 使用
def example_profile() -> None:
with profile("数据处理") as logs:
logs.append("正在处理数据...")
data: list[int] = list(range(10000))
logs.append(f"处理了 {len(data)} 条数据")嵌套上下文管理器
python
from contextlib import contextmanager
from typing import Generator
from dataclasses import dataclass
import time
@dataclass
class DatabaseConnection:
"""模拟数据库连接"""
host: str
connected: bool = False
def connect(self) -> None:
print(f"连接数据库：{self.host}")
self.connected = True
def disconnect(self) -> None:
print(f"断开数据库：{self.host}")
self.connected = False
def execute(self, query: str) -> list[dict[str, str]]:
if not self.connected:
raise RuntimeError("数据库未连接")
print(f"执行查询：{query}")
return [{"id": "1", "name": "Alice"}]
@contextmanager
def database(host: str) -> Generator[DatabaseConnection, None, None]:
"""数据库连接上下文管理器"""
conn: DatabaseConnection = DatabaseConnection(host)
try:
conn.connect()
yield conn
finally:
conn.disconnect()
@contextmanager
def transaction(conn: DatabaseConnection) -> Generator[None, None, None]:
"""事务上下文管理器"""
print("开始事务")
try:
yield
print("提交事务")
except Exception as e:
print(f"回滚事务：{e}")
raise
@contextmanager
def timer_context(name: str) -> Generator[None, None, None]:
"""计时上下文管理器"""
start: float = time.time()
try:
yield
finally:
print(f"{name} 耗时：{time.time() - start:.4f} 秒")
# 使用嵌套上下文管理器
def complex_operation() -> None:
"""复杂操作：数据库 + 事务 + 计时"""
with timer_context("数据库操作"):
with database("localhost:5432") as conn:
with transaction(conn):
users: list[dict[str, str]] = conn.execute("SELECT * FROM users")
print(f"查询结果：{users}")
# Python 3.10+ 单行写法
def complex_operation_modern() -> None:
"""Python 3.10+ 单行写法"""
with timer_context("数据库操作"), \
database("localhost:5432") as conn, \
transaction(conn):
users: list[dict[str, str]] = conn.execute("SELECT * FROM users")
print(f"查询结果：{users}")第六步：实际应用
文件处理助手
python
from pathlib import Path
from typing import Generator, Any
from contextlib import contextmanager
import json
@contextmanager
def safe_write(
filepath: str | Path,
encoding: str = "utf-8",
backup: bool = True
) -> Generator[Any, None, None]:
"""
安全写入文件的上下文管理器。
Args:
filepath: 文件路径
encoding: 文件编码
backup: 是否备份原文件
Yields:
文件对象
"""
path: Path = Path(filepath)
backup_path: Path | None = None
# 备份原文件
if backup and path.exists():
backup_path = path.with_suffix(path.suffix + ".bak")
backup_path.write_text(path.read_text(encoding=encoding), encoding=encoding)
try:
with open(path, "w", encoding=encoding) as f:
yield f
print(f"✅ 文件保存成功：{path}")
except Exception as e:
# 恢复备份
if backup_path and backup_path.exists():
path.write_text(backup_path.read_text(encoding=encoding), encoding=encoding)
backup_path.unlink()
print(f"❌ 文件保存失败：{e}")
raise
finally:
# 清理备份
if backup_path and backup_path.exists():
backup_path.unlink()
@contextmanager
def json_file(filepath: str | Path, default: Any = None) -> Generator[Any, None, None]:
"""
JSON 文件处理上下文管理器。
Args:
filepath: 文件路径
default: 默认值（文件不存在时）
Yields:
可修改的数据对象，退出时自动保存
"""
path: Path = Path(filepath)
data: Any
# 读取数据
if path.exists():
try:
data = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = default if default is not None else {}
else:
data = default if default is not None else {}
# 提供可修改的数据
try:
yield data
finally:
# 自动保存
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
# 使用示例
def example_file_helpers() -> None:
"""文件处理助手示例"""
# 安全写入
with safe_write("config.json") as f:
json.dump({"version": "1.0"}, f)
# JSON 配置文件
with json_file("app_config.json", default={"theme": "light"}) as config:
config["theme"] = "dark"
config["language"] = "zh-CN"
# 退出时自动保存性能监控工具
python
import time
from contextlib import contextmanager
from typing import Generator, Callable, Any
from dataclasses import dataclass, field
from functools import wraps
@dataclass
class PerformanceMetrics:
"""性能指标"""
name: str
start_time: float = 0.0
end_time: float = 0.0
duration: float = 0.0
success: bool = True
error: str | None = None
calls: int = 0
total_time: float = 0.0
def __str__(self) -> str:
status: str = "✅" if self.success else "❌"
return f"{status} {self.name}: {self.duration:.4f}s"
class PerformanceMonitor:
"""性能监控器"""
def __init__(self) -> None:
self.metrics: dict[str, PerformanceMetrics] = {}
@contextmanager
def measure(
self,
name: str,
print_result: bool = True
) -> Generator[PerformanceMetrics, None, None]:
"""测量代码块执行时间"""
metric: PerformanceMetrics = PerformanceMetrics(name)
metric.start_time = time.time()
try:
yield metric
metric.success = True
except Exception as e:
metric.success = False
metric.error = str(e)
raise
finally:
metric.end_time = time.time()
metric.duration = metric.end_time - metric.start_time
# 更新统计
if name in self.metrics:
existing: PerformanceMetrics = self.metrics[name]
existing.calls += 1
existing.total_time += metric.duration
else:
metric.calls = 1
metric.total_time = metric.duration
self.metrics[name] = metric
if print_result:
print(metric)
def summary(self) -> str:
"""生成性能报告"""
lines: list[str] = ["=" * 50, "性能报告", "=" * 50]
for name, metric in self.metrics.items():
avg_time: float = metric.total_time / metric.calls if metric.calls > 0 else 0
lines.append(
f"{name}: "
f"调用 {metric.calls} 次, "
f"总时间 {metric.total_time:.4f}s, "
f"平均 {avg_time:.4f}s"
)
lines.append("=" * 50)
return "\n".join(lines)
# 装饰器版本
def measure_time(name: str | None = None) -> Callable:
"""测量函数执行时间的装饰器"""
def decorator(func: Callable) -> Callable:
metric_name: str = name or func.__name__
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
start: float = time.time()
try:
result: Any = func(*args, **kwargs)
duration: float = time.time() - start
print(f"✅ {metric_name}: {duration:.4f}s")
return result
except Exception as e:
duration: float = time.time() - start
print(f"❌ {metric_name}: {duration:.4f}s (错误: {e})")
raise
return wrapper
return decorator
# 使用示例
monitor: PerformanceMonitor = PerformanceMonitor()
def example_performance() -> None:
"""性能监控示例"""
# 上下文管理器方式
with monitor.measure("数据处理"):
data: list[int] = [x * 2 for x in range(100000)]
with monitor.measure("排序"):
sorted_data: list[int] = sorted(data, reverse=True)
# 装饰器方式
@measure_time("复杂计算")
def complex_calculation(n: int) -> int:
return sum(i * i for i in range(n))
result: int = complex_calculation(100000)
# 打印报告
print("\n" + monitor.summary())数据库事务管理
python
from contextlib import contextmanager
from typing import Generator, Any, Protocol
from dataclasses import dataclass, field
class DatabaseBackend(Protocol):
"""数据库后端协议"""
def execute(self, query: str, params: tuple = ()) -> list[dict[str, Any]]:
...
def commit(self) -> None:
...
def rollback(self) -> None:
...
def close(self) -> None:
...
@dataclass
class MockDatabase:
"""模拟数据库"""
connected: bool = False
in_transaction: bool = False
def execute(self, query: str, params: tuple = ()) -> list[dict[str, Any]]:
print(f"执行 SQL：{query} 参数：{params}")
return [{"id": 1, "name": "Alice"}]
def commit(self) -> None:
print("提交事务")
self.in_transaction = False
def rollback(self) -> None:
print("回滚事务")
self.in_transaction = False
def close(self) -> None:
print("关闭连接")
self.connected = False
@contextmanager
def database_connection(
host: str,
port: int = 5432
) -> Generator[MockDatabase, None, None]:
"""数据库连接上下文管理器"""
db: MockDatabase = MockDatabase(connected=True)
print(f"连接数据库：{host}:{port}")
try:
yield db
finally:
db.close()
@contextmanager
def transaction(db: DatabaseBackend) -> Generator[None, None, None]:
"""事务上下文管理器"""
print("开始事务")
db.in_transaction = True
try:
yield
db.commit()
except Exception as e:
print(f"事务失败：{e}")
db.rollback()
raise
@contextmanager
def savepoint(
db: DatabaseBackend,
name: str = "savepoint"
) -> Generator[None, None, None]:
"""保存点上下文管理器"""
print(f"创建保存点：{name}")
db.execute(f"SAVEPOINT {name}")
try:
yield
print(f"释放保存点：{name}")
db.execute(f"RELEASE SAVEPOINT {name}")
except Exception as e:
print(f"回滚到保存点：{name}")
db.execute(f"ROLLBACK TO SAVEPOINT {name}")
raise
# 使用示例
def example_database() -> None:
"""数据库事务示例"""
# 基本事务
with database_connection("localhost") as db:
with transaction(db):
db.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
db.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
print("\n--- 带保存点的事务 ---\n")
# 带保存点的事务
with database_connection("localhost") as db:
with transaction(db):
db.execute("INSERT INTO users (name) VALUES (?)", ("Charlie",))
try:
with savepoint(db, "sp1"):
db.execute("INSERT INTO users (name) VALUES (?)", ("David",))
# 模拟错误
raise ValueError("模拟错误")
except ValueError:
print("捕获错误，继续事务")
db.execute("INSERT INTO users (name) VALUES (?)", ("Eve",))
if __name__ == "__main__":
example_database()关键代码说明：
| 代码 | 含义 | 为什么这样写 |
|---|---|---|
@contextmanager + yield | 用生成器实现上下文管理器 | yield 前的代码相当于 __enter__，yield 后相当于 __exit__，代码更线性易读 |
with database_connection(...) as db: + with transaction(db): | 嵌套 with 语句 | 连接和事务是两个独立资源，分别管理各自的生命周期，嵌套体现层级关系 |
except Exception: db.rollback(); raise | 事务异常时回滚并重新抛出 | 数据一致性要求：任何错误都必须回滚；raise 确保调用方仍能感知错误 |
with savepoint(db, "sp1"): 嵌套在 transaction 内 | 保存点嵌套在事务中 | 保存点允许部分回滚而不影响外层事务，实现更细粒度的错误恢复 |
except ValueError: print("捕获错误，继续事务") | 捕获保存点内的错误后继续 | 保存点回滚后事务仍然有效，可以继续执行后续操作 |
L2 实践层：最佳实践
with 语句最佳实践
推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 总是用 with 处理资源 | 确保资源释放，避免泄漏 | with open(...) as f: |
| 一个 with 管一个资源 | 职责清晰，便于调试 | 分开管理，而非混在一起 |
使用 as 绑定对象 | 代码块内可访问资源 | with open(...) as f: |
| Python 3.10+ 用括号语法 | 多个管理器更清晰 | with (a, b, c): |
| 自定义管理器用 @contextmanager | 代码更简洁，易理解 | 比 __enter__/__exit__ 更简单 |
多个上下文管理器的写法
python
# ✅ 推荐：Python 3.10+ 括号语法（清晰易读）
def process_files_modern(src: str, dst: str) -> None:
with (
open(src, "r") as fin,
open(dst, "w") as fout,
):
fout.write(fin.read())
# ✅ 推荐：Python 3.9 及更早（反斜杠续行）
def process_files_legacy(src: str, dst: str) -> None:
with open(src, "r") as fin, \
open(dst, "w") as fout:
fout.write(fin.read())
# ❌ 不推荐：嵌套 with（层数太多）
def process_files_nested(src: str, dst: str) -> None:
with open(src, "r") as fin:
with open(dst, "w") as fout:
fout.write(fin.read())
# 问题：3 个资源就嵌套 3 层，代码缩进太深with vs try-finally 的选择
┌─────────────────────────────────────────────────────────────────┐
│ with vs try-finally 选择指南 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 使用 with 的场景： │
│ ───────────────────────────────────────────────────────── │
│ • 有现成的上下文管理器（文件、锁、连接） │
│ • 只需要资源管理，不需要复杂异常处理 │
│ • 代码简洁优先 │
│ │
│ 使用 try-finally 的场景： │
│ ───────────────────────────────────────────────────────── │
│ • 没有现成的上下文管理器 │
│ • 需要复杂的异常处理逻辑 │
│ • 需要在 finally 中做多种清理操作 │
│ │
│ ───────────────────────────────────────────────────────── │
│ │
│ 对比示例： │
│ │
│ # 方式 1：with（推荐） │
│ with open("file.txt") as f: │
│ content = f.read() │
│ # 自动关闭，简洁 │
│ │
│ # 方式 2：try-finally（需要更多代码） │
│ f = None │
│ try: │
│ f = open("file.txt") │
│ content = f.read() │
│ finally: │
│ if f: │
│ f.close() │
│ # 手动关闭，繁琐 │
│ │
└─────────────────────────────────────────────────────────────────┘反模式对比
反模式 1：不用 with 管理资源
python
# ❌ 不推荐：手动管理资源
def read_file_bad(filepath: str) -> str:
f = open(filepath, "r")
content = f.read()
f.close() # 如果 read() 抛异常，这行不执行
return content
# 问题：read() 抛异常时，close() 不执行，资源泄漏python
# ✅ 推荐：用 with 管理
def read_file_good(filepath: str) -> str:
with open(filepath, "r") as f:
content = f.read()
return content # 无论是否异常，文件都会关闭反模式 2：在 with 块外使用资源
python
# ❌ 不推荐：with 块外使用资源
def process_file_bad(filepath: str) -> str:
with open(filepath, "r") as f:
pass # 没有在块内使用
return f.read() # f 已关闭，这里会失败
# 问题：with 块结束时资源已释放，外部使用无效python
# ✅ 推荐：在 with 块内完成所有操作
def process_file_good(filepath: str) -> str:
with open(filepath, "r") as f:
content = f.read()
processed = content.upper() # 所有操作在块内完成
return processed反模式 3：嵌套太深
python
# ❌ 不推荐：多个资源嵌套
def complex_operation_bad() -> None:
with open("input.txt") as fin:
with open("output.txt") as fout:
with acquire_lock(lock):
with timer("operation"):
fout.write(fin.read())
# 问题：缩进 4 层，难以阅读python
# ✅ 推荐：合并多个管理器
def complex_operation_good() -> None:
with (
open("input.txt") as fin,
open("output.txt") as fout,
acquire_lock(lock),
timer("operation"),
):
fout.write(fin.read())
# 清晰：所有资源在一行管理，缩进统一反模式汇总表
| 反模式 | 错误代码 | 问题 | 正确做法 |
|---|---|---|---|
| 不用 with | f=open(); f.read(); f.close() | 异常时资源泄漏 | with open() as f: |
| 块外使用 | with ... as f: pass; f.read() | 资源已释放 | 所有操作在块内 |
| 嵌套太深 | with a: with b: with c: | 缩进太深难读 | with a, b, c: |
| 混用返回值 | with ...: return f | 返回已关闭对象 | 返回内容而非对象 |
| 忽略异常信息 | __exit__ 不检查 exc_type | 无法处理特定异常 | 根据异常类型决定返回值 |
适用场景
上下文管理器使用场景指南
| 场景 | 推荐程度 | 推荐做法 | 原因 |
|---|---|---|---|
| 文件操作 | ✅ 必须 | with open() | 文件必须关闭，with 保证执行 |
| 数据库连接 | ✅ 必须 | with connection() | 连接必须释放，防止泄漏 |
| 锁管理 | ✅ 必须 | with lock: | 锁必须释放，防止死锁 |
| 事务管理 | ✅ 必须 | with transaction() | 事务必须提交或回滚 |
| 临时状态 | ✅ 推荐 | 自定义管理器 | 进入/退出时切换状态 |
| 计时/性能 | ✅ 推荐 | @contextmanager | 进入开始计时，退出结束 |
| 简单变量 | ❌ 不需要 | 不用管理器 | 变量不需要"释放" |
L3 专家层：底层原理
enter/exit 协议
Python 如何实现上下文管理器
┌─────────────────────────────────────────────────────────────────┐
│ 上下文管理器协议 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 上下文管理器必须实现两个方法： │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ __enter__(self) │ │
│ │ • 进入上下文时调用 │ │
│ │ • 返回值绑定到 as 后的变量 │ │
│ │ • 可以返回 self 或其他对象 │ │
│ │ • 用于获取资源、设置状态 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ __exit__(self, exc_type, exc_val, exc_tb) │ │
│ │ • 退出上下文时调用（总是执行） │ │
│ │ • exc_type: 异常类型（无异常时为 None） │ │
│ │ • exc_val: 异常对象（无异常时为 None） │ │
│ │ • exc_tb: traceback 对象（无异常时为 None） │ │
│ │ • 返回 True: 抑制异常，继续执行 │ │
│ │ • 返回 False: 异常继续传播 │ │
│ │ • 用于释放资源、清理状态 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ ───────────────────────────────────────────────────────── │
│ │
│ with 语句执行流程： │
│ │
│ 1. 创建上下文管理器对象 │
│ 2. 调用 __enter__()，结果绑定到变量 │
│ 3. 执行 with 块内的代码 │
│ 4. 调用 __exit__()，传入异常信息（如果有） │
│ 5. 根据 __exit__ 返回值决定异常是否传播 │
│ │
└─────────────────────────────────────────────────────────────────┘exit 参数详解
python
# 演示 __exit__ 的参数
from typing import Optional, Any
class DetailedContext:
"""详细的上下文管理器，展示 __exit__ 参数"""
def __enter__(self) -> "DetailedContext":
print("进入上下文")
return self
def __exit__(
self,
exc_type: Optional[type], # 异常类型（如 ValueError）
exc_val: Optional[BaseException], # 异常实例
exc_tb: Optional[Any] # traceback 对象
) -> bool:
print(f"退出上下文")
print(f" exc_type: {exc_type}") # None 或异常类
print(f" exc_val: {exc_val}") # None 或异常对象
print(f" exc_tb: {exc_tb}") # None 或 traceback
if exc_type is not None:
print(f" 发生异常：{exc_type.__name__}: {exc_val}")
# 返回 True 可以抑制异常
# 返回 False 让异常继续传播
return False # 不抑制异常
return False
# 测试：无异常情况
def test_no_exception() -> None:
with DetailedContext():
print("执行代码块")
# 输出：exc_type, exc_val, exc_tb 都是 None
# 测试：有异常情况
def test_with_exception() -> None:
try:
with DetailedContext():
print("执行代码块")
raise ValueError("测试异常")
except ValueError:
print("捕获到异常")
# 输出：exc_type=ValueError, exc_val=ValueError实例, exc_tb=traceback抑制异常的实现
python
# ✅ 抑制特定异常的上下文管理器
from typing import Optional, Any
class SuppressExceptions:
"""抑制指定类型的异常"""
def __init__(self, *exceptions: type) -> None:
self.exceptions = exceptions
def __enter__(self) -> "SuppressExceptions":
return self
def __exit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[Any]
) -> bool:
# 如果发生了异常，且异常类型在指定范围内
if exc_type and issubclass(exc_type, self.exceptions):
print(f"抑制异常：{exc_type.__name__}: {exc_val}")
return True # 返回 True 表示异常已处理，不传播
return False # 其他异常正常传播
# 使用
def example_suppress() -> None:
with SuppressExceptions(ZeroDivisionError, ValueError):
result = 10 / 0 # ZeroDivisionError 被抑制
print("这行不会执行（异常发生在前面）")
print("继续执行，异常被抑制")
# 等价于 contextlib.suppress
from contextlib import suppress
with suppress(ZeroDivisionError):
result = 10 / 0
print("异常被抑制")contextlib 模块深入
Python 如何实现 @contextmanager
┌─────────────────────────────────────────────────────────────────┐
│ @contextmanager 实现原理 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ @contextmanager 装饰器将生成器函数转换为上下文管理器： │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ @contextmanager │ │
│ │ def my_context(): │ │
│ │ # yield 前的代码 = __enter__ │ │
│ │ setup() │ │
│ │ yield resource │ │
│ │ # yield 后的代码 = __exit__ │ │
│ │ cleanup() │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ 内部实现： │
│ ───────────────────────────────────────────────────────── │
│ 1. 装饰器创建一个 ContextManager 类 │
│ 2. 类的 __enter__ 调用生成器的 next()，到达 yield │
│ 3. yield 的值作为 __enter__ 的返回值 │
│ 4. 类的 __exit__ 继续执行生成器（next() 或 throw()） │
│ 5. 如果 with 块有异常，用 throw() 传入生成器 │
│ 6. 生成器执行完毕后，清理工作完成 │
│ │
│ ───────────────────────────────────────────────────────── │
│ │
│ 关键点： │
│ • yield 前的代码在进入时执行 │
│ • yield 后的代码在退出时执行（包括异常情况） │
│ • try-finally 包裹 yield 确保清理 │
│ • 异常会通过 throw() 传回生成器 │
│ │
└─────────────────────────────────────────────────────────────────┘contextlib 常用工具详解
python
from contextlib import (
contextmanager,
suppress,
redirect_stdout,
redirect_stderr,
closing,
nullcontext,
ExitStack,
)
import io
import sys
from typing import Generator, Any, TextIO
# 1. suppress：抑制异常
def demo_suppress() -> None:
"""suppress 的等价实现"""
# contextlib.suppress
with suppress(FileNotFoundError):
import os
os.remove("不存在的文件.txt")
# 文件不存在也不会报错
# 等价于
try:
os.remove("不存在的文件.txt")
except FileNotFoundError:
pass
# 2. redirect_stdout/redirect_stderr：重定向输出
def demo_redirect() -> None:
"""捕获标准输出"""
output: io.StringIO = io.StringIO()
with redirect_stdout(output):
print("这行被捕获")
print("这行也被捕获")
captured: str = output.getvalue()
print(f"捕获的内容：{captured}")
# 应用场景：测试输出、日志捕获
# 3. closing：确保调用 close() 方法
def demo_closing() -> None:
"""closing 确保资源关闭"""
# 任何有 close() 方法的对象都可以用 closing
class MyResource:
def close(self) -> None:
print("资源已关闭")
with closing(MyResource()) as resource:
print("使用资源")
# 退出时自动调用 resource.close()
# 4. nullcontext：空上下文管理器（用于条件性管理）
def demo_nullcontext() -> None:
"""条件性使用上下文管理器"""
use_lock = False
lock = None # 模拟锁
# 如果不需要锁，用 nullcontext 代替
ctx = lock if use_lock else nullcontext()
with ctx:
print("执行操作")
# 如果 use_lock=False，这里是空操作
# 5. ExitStack：动态管理多个上下文
def demo_exit_stack() -> None:
"""动态数量的上下文管理器"""
from contextlib import ExitStack
files: list[str] = ["file1.txt", "file2.txt", "file3.txt"]
# 数量不固定时，用 ExitStack
with ExitStack() as stack:
handles: list[TextIO] = [
stack.enter_context(open(f, "w"))
for f in files
]
# 所有文件都在 stack 中注册
for handle in handles:
handle.write("content")
# 退出时，所有文件自动关闭性能考量
上下文管理器的性能特性
┌─────────────────────────────────────────────────────────────────┐
│ 上下文管理器性能分析 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 关键结论： │
│ │
│ 1. with 语句本身开销很低 │
│ • 只是调用 __enter__ 和 __exit__ │
│ • 比手动 try-finally 更高效（代码更少） │
│ │
│ 2. @contextmanager vs 类实现 │
│ • 性能差异很小 │
│ • @contextmanager 更简洁，推荐用于简单场景 │
│ • 类实现更灵活，推荐用于复杂场景 │
│ │
│ 3. 嵌套管理器的开销 │
│ • 每个管理器调用一次 __enter__/__exit__ │
│ • 多个管理器开销累加，但通常可忽略 │
│ • 合并写法 vs 嵌套写法，性能相同 │
│ │
│ 4. 资源获取/释放才是主要开销 │
│ • 文件打开/关闭、连接建立/断开 │
│ • 上下文管理器本身不是瓶颈 │
│ │
└─────────────────────────────────────────────────────────────────┘性能对比测试
python
import timeit
import time
from contextlib import contextmanager
from typing import Generator
# 测试：@contextmanager vs 类实现
@contextmanager
def timer_generator() -> Generator[None, None, None]:
"""生成器版本的计时器"""
start = time.time()
yield
end = time.time()
print(f"耗时：{end - start:.6f}s")
class TimerClass:
"""类版本的计时器"""
def __enter__(self) -> "TimerClass":
self.start = time.time()
return self
def __exit__(self, *args) -> bool:
end = time.time()
print(f"耗时：{end - self.start:.6f}s")
return False
# 性能测试（仅测试管理器本身，不含实际操作）
def test_generator() -> None:
with timer_generator():
pass # 无操作
def test_class() -> None:
with TimerClass():
pass # 无操作
# 结果：差异很小（约 1-2 微秒）
# 生成器版本稍慢，因为需要创建生成器对象
# 但实际使用中，资源操作才是主要开销设计动机
Python 为什么这样设计上下文管理器
┌─────────────────────────────────────────────────────────────────┐
│ 上下文管理器设计动机 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 设计选择 原因 │
│ ───────────────────────────────────────────────────────── │
│ │
│ __enter__/__exit__ 协议 │
│ • 用方法而非特殊语法 │
│ • 灵活性高，可以在方法中做任何操作 │
│ • __enter__ 返回值绑定到 as 变量 │
│ • __exit__ 接收异常信息，可选择抑制 │
│ │
│ with 语句 │
│ • 语法简洁，一眼看出资源范围 │
│ • 比嵌套 try-finally 更清晰 │
│ • 支持多个管理器（Python 3.1+） │
│ • 括号语法（Python 3.10+）更美观 │
│ │
│ @contextmanager 装饰器 │
│ • 用生成器实现，代码更线性 │
│ • yield 前后明确分离进入/退出逻辑 │
│ • 简单场景不需要定义类 │
│ • Python 2.5 引入，简化常见用法 │
│ │
│ contextlib 模块 │
│ • 提供常用工具（suppress、redirect_stdout） │
│ • ExitStack 处理动态数量管理器 │
│ • 减少重复代码，提高可复用性 │
│ │
└─────────────────────────────────────────────────────────────────┘与其他语言的对比
| 特性 | Python | Java | Go | C++ |
|---|---|---|---|---|
| 资源管理语法 | with | try-with-resources | defer | RAII（析构函数） |
| 协议方法 | __enter__/__exit__ | AutoCloseable.close() | 函数调用 | 析构函数 |
| 异常处理 | __exit__ 可抑制 | 无法抑制 | 无法抑制 | 无法抑制 |
| 简洁实现 | @contextmanager | 无 | 无 | 无 |
| 多资源 | with a, b: | try(a; b) | 多个 defer | 多个对象 |
知识关联
┌─────────────────────────────────────────────────────────────────┐
│ 上下文管理器知识关联图 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────┐ │
│ │ 魔术方法 │ │
│ │ __enter__ │ │
│ │ __exit__ │ │
│ └───────────────────┘ │
│ │ │
│ ↓ │
│ ┌───────────────────┐ │
│ │ with 语句 │ │
│ │ (使用入口) │ │
│ └───────────────────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │文件管理 │ │ 锁管理 │ │事务管理 │ │
│ │open() │ │Lock │ │Transaction│ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌───────────────────────────────────────┐ │
│ │ contextlib 模块 │ │
│ │ • @contextmanager │ │
│ │ • suppress │ │
│ │ • redirect_stdout │ │
│ │ • ExitStack │ │
│ └───────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌───────────────────┐ │
│ │ 生成器 │ │
│ │ yield 关键字 │ │
│ └───────────────────┘ │
│ │ │
│ ↓ │
│ ┌───────────────────┐ │
│ │ 异常处理 │ │
│ │ __exit__ 返回值 │ │
│ │ 决定异常传播 │ │
│ └───────────────────┘ │
│ │
│ 前置知识：类与对象、魔术方法、生成器 │
│ 进阶扩展：自定义上下文管理器、ExitStack 动态管理 │
│ │
└─────────────────────────────────────────────────────────────────┘本章小结
┌─────────────────────────────────────────────────────────────┐
│ 上下文管理器 知识要点 │
├─────────────────────────────────────────────────────────────┤
│ │
│ with 语句： │
│ ✓ 自动管理资源 │
│ ✓ 确保资源释放 │
│ ✓ 支持多个上下文管理器 │
│ │
│ 上下文管理器协议： │
│ ✓ __enter__：进入上下文 │
│ ✓ __exit__：离开上下文 │
│ ✓ __exit__ 返回 True 可抑制异常 │
│ │
│ contextlib： │
│ ✓ @contextmanager 装饰器 │
│ ✓ suppress() 抑制异常 │
│ ✓ redirect_stdout() 重定向输出 │
│ │
│ 应用场景： │
│ ✓ 文件操作 │
│ ✓ 数据库连接 │
│ ✓ 锁的获取和释放 │
│ ✓ 性能测量 │
│ │
└─────────────────────────────────────────────────────────────┘恭喜完成错误与异常章节！ 你已经掌握了 Python 异常处理的完整知识体系。