【一起学 Python】第 64 天:上下文管理器 with 语句的深度解析

# ❌ 繁琐的 try-finallyf = open("data.txt", "r")try:    content = f.read()finally:    f.close()# ✅ 优雅的 with 语句with open("data.txt", "r") as f:    content = f.read()# 自动关闭文件，即使发生异常

# with 语句with expression as variable:    block# 等价于manager = expressionvariable = manager.__enter__()try:    blockfinally:    manager.__exit__(*sys.exc_info())

class FileManager:    def __init__(self, filename, mode):        self.filename = filename        self.mode = mode        self.file = None    def __enter__(self):        print("进入 with 块")        self.file = open(self.filename, self.mode)        return self.file  # 返回给 as 后的变量    def __exit__(self, exc_type, exc_val, exc_tb):        print("离开 with 块")        if self.file:            self.file.close()        # 返回 None 或 False，异常会继续传播        return False# 使用with FileManager("data.txt", "r") as f:    content = f.read()

def __exit__(self, exc_type, exc_val, exc_tb):    """    exc_type: 异常类型（如 ValueError）    exc_val:  异常值（如错误信息）    exc_tb:   异常堆栈跟踪对象    """    if exc_type:        print(f"发生异常：{exc_type.__name__}: {exc_val}")    # 返回 True 抑制异常，返回 False 或 None 让异常继续传播

class SuppressError:    def __enter__(self):        return self    def __exit__(self, exc_type, exc_val, exc_tb):        print("捕获并抑制异常")        return True  # 抑制异常with SuppressError():    raise ValueError("这个异常不会传播")print("程序继续执行")  # 会执行

import loggingclass LogErrors:    def __enter__(self):        return self    def __exit__(self, exc_type, exc_val, exc_tb):        if exc_type:            logging.error(f"{exc_type.__name__}: {exc_val}")        return False  # 让异常继续传播with LogErrors():    raise ValueError("记录日志后继续传播")

from contextlib import contextmanager@contextmanagerdef open_file(filename, mode):    f = open(filename, mode)    try:        yield f  # yield 前的代码是 __enter__，之后是 __exit__    finally:        f.close()# 使用with open_file("data.txt", "r") as f:    content = f.read()

from contextlib import contextmanager@contextmanagerdef transaction(db):    db.begin()    try:        yield db        db.commit()    except Exception:        db.rollback()        raise  # 重新抛出异常# 使用with transaction(database) as db:    db.execute("INSERT INTO users ...")

from contextlib import closingimport urllib.requestwith closing(urllib.request.urlopen("http://example.com")) as response:    content = response.read()

from contextlib import suppresswith suppress(FileNotFoundError):    os.remove("不存在的文件.txt")# 不会抛出异常# 等价于try:    os.remove("不存在的文件.txt")except FileNotFoundError:    pass

from typing import Optionalfrom contextlib import contextmanagerclass DatabaseConnection:    """数据库连接上下文管理器"""    def __init__(self, host: str, port: int, database: str):        self.host = host        self.port = port        self.database = database        self.connection = None    def __enter__(self):        print(f"连接到 {self.host}:{self.port}/{self.database}")        # 模拟连接        self.connection = {"host": self.host, "connected": True}        return self.connection    def __exit__(self, exc_type, exc_val, exc_tb):        print("关闭数据库连接")        if self.connection:            self.connection["connected"] = False        # 返回 False，让异常继续传播        return False    def execute(self, sql: str):        if not self.connection or not self.connection["connected"]:            raise RuntimeError("数据库未连接")        print(f"执行 SQL: {sql}")        return []# 使用with DatabaseConnection("localhost", 3306, "mydb") as db:    db.execute("SELECT * FROM users")# 输出：# 连接到 localhost:3306/mydb# 执行 SQL: SELECT * FROM users# 关闭数据库连接

import timefrom contextlib import contextmanager@contextmanagerdef timer(name: str = "操作"):    """计时上下文管理器"""    start = time.time()    try:        yield    finally:        elapsed = time.time() - start        print(f"{name} 耗时：{elapsed:.4f}秒")# 使用with timer("数据处理"):    time.sleep(1)    result = sum(range(1000000))with timer("文件读取"):    with open("data.txt", "w") as f:        f.write("test")

import osimport shutilfrom pathlib import Pathfrom contextlib import contextmanager@contextmanagerdef temporary_directory(path: str):    """创建临时目录，退出时自动删除"""    dir_path = Path(path)    dir_path.mkdir(parents=True, exist_ok=True)    print(f"创建临时目录：{dir_path}")    try:        yield dir_path    finally:        print(f"删除临时目录：{dir_path}")        shutil.rmtree(dir_path, ignore_errors=True)# 使用with temporary_directory("temp/data") as tmp_dir:    # 在临时目录中操作    (tmp_dir / "file.txt").write_text("内容")    print(f"临时文件：{tmp_dir / 'file.txt'}")# 退出后目录自动删除

# 返回 True：抑制异常class SuppressAll:    def __exit__(self, *args):        return Truewith SuppressAll():    raise ValueError("不会传播")print("继续执行")  # 会执行# 返回 False/None：传播异常class PropagateError:    def __exit__(self, *args):        return Falsewith PropagateError():    raise ValueError("会传播")print("不会执行")  # 不会执行

# 传统写法with open("input.txt") as f_in:    with open("output.txt", "w") as f_out:        f_out.write(f_in.read())# Python 3.10+ 括号写法with (    open("input.txt") as f_in,    open("output.txt", "w") as f_out):    f_out.write(f_in.read())

class BadExit:    def __exit__(self, *args):        raise RuntimeError("清理时出错")with BadExit():    raise ValueError("原始错误")# 最终只看到 RuntimeError，原始错误丢失

# ❌ 错误@contextmanagerdef bad_context():    setup()    yield    yield  # 第二次 yield 会报错    cleanup()# ✅ 正确@contextmanagerdef good_context():    setup()    try:        yield    finally:        cleanup()