python之contextmanager

一、contextmanager有什么用？

contextmanager是 Python 标准库contextlib模块提供的同步上下文管理器装饰器，核心作用是简化同步上下文管理器的实现。

在没有contextmanager之前，实现一个上下文管理器需要手动定义类，并实现__enter__()和__exit__()两个魔法方法；而使用contextmanager，只需装饰一个同步生成器函数，即可快速封装出上下文管理器，无需手动实现魔法方法，代码更简洁优雅。

二、适用场景

当你在同步代码中需要管理「需要先获取、使用后必须释放 / 清理」的资源时，contextmanager是最优选择，

典型场景包括：

• 同步文件操作：普通文件的打开 / 关闭，自动释放文件句柄（Python 内置open()本身就是上下文管理器，此处用于演示封装思想）
• 同步数据库连接：如pymysql操作 MySQL，自动建立 / 关闭数据库连接
• 同步锁 / 资源占用：如threading.Lock，自动获取 / 释放锁，避免死锁
• 临时资源创建 / 清理：如临时目录创建、临时文件写入后自动删除、日志上下文切换等

简单来说：同步场景 + 资源需要自动管理（获取 + 释放），就用contextmanager。

三、核心使用规则

@contextmanager装饰的同步生成器函数，必须满足固定结构：

• 函数用def定义（同步函数，无async关键字）
• 函数内部包含且仅包含一个yield关键字（用于分割「资源获取」和「资源释放」逻辑）
• 代码结构分为三部分：
  1. yield 之前：对应上下文管理器的__enter__()方法，负责获取 / 初始化资源，执行时机是进入with语句块时
  2. yield 关键字：将需要使用的资源「返回」给with语句的as变量（可返回单个资源，也可返回None），同时暂停函数执行，交出控制权给with语句块
  3. yield 之后：对应上下文管理器的__exit__()方法，负责释放 / 清理资源，执行时机是离开with语句块时（无论代码块正常结束还是发生异常，都会执行）
• 推荐将yield之后的清理逻辑放在finally块中，确保资源释放的可靠性（即使中间代码抛出异常，清理逻辑也能执行）

四、代码示例

我们实现一个「同步文件读写工具」，使用contextmanager封装普通文件的打开 / 关闭，确保文件句柄自动释放，无需手动调用close()，与之前的异步示例形成对比，更易理解。

# 1. 导入必要模块（contextmanager是Python内置，无需额外安装依赖）fromcontextlibimportcontextmanager# 2. 使用@contextmanager装饰同步生成器，实现同步上下文管理器@contextmanagerdeffile_manager(file_path:str,mode:str="r",encoding:str="utf-8"):""" 同步文件管理器：自动打开/关闭文件，封装同步文件操作的资源管理 :param file_path: 文件路径 :param mode: 文件打开模式（r/w/a等） :param encoding: 文件编码 """# 初始化文件句柄file_handle=Nonetry:# -------- 第一步：yield之前（对应__enter__()）：获取/初始化资源 --------file_handle=open(file_path,mode=mode,encoding=encoding)print(f"✅ 同步打开文件：{file_path}（模式：{mode}）")# -------- 第二步：yield关键字：返回资源给with语句，暂停函数执行 --------yieldfile_handle# 将文件句柄传递给with ... as 后的变量# -------- 第三步：yield之后（对应__exit__()）：释放/清理资源（finally确保必执行） --------finally:iffile_handle:file_handle.close()print(f"❌ 同步关闭文件：{file_path}")# 3. 使用同步上下文管理器（配合with语句）defmain():# 测试文件路径test_file_path="test_sync.txt"# 示例1：同步写入文件内容withfile_manager(test_file_path,mode="w")asf:f.write("Hello, contextmanager!\n")f.write("这是同步文件写入测试\n")# 示例2：同步读取文件内容withfile_manager(test_file_path,mode="r")asf:content=f.read()print("\n📄 文件内容：")print(content)# 4. 运行同步主函数if__name__=="__main__":main()

关键解析

• 进入with语句块时，自动执行yield之前的open()逻辑，打开文件并获取文件句柄
• with语句块内操作文件句柄，执行完成后自动离开代码块，触发finally中的close()逻辑，关闭文件
• 即使文件写入 / 读取过程中抛出异常（例如手动添加raise Exception(“测试异常”)），finally块中的清理逻辑依然会执行，确保文件句柄被释放，不会造成资源泄露

异常处理

@contextmanagerdeffile_manager(file_path:str,mode:str="r",encoding:str="utf-8"):file_handle=Nonetry:file_handle=open(file_path,mode=mode,encoding=encoding)print(f"✅ 同步打开文件：{file_path}（模式：{mode}）")yieldfile_handle# 此处抛出的异常会被传递到with代码块exceptFileNotFoundErrorase:print(f"❌ 错误：文件不存在 -{e}")raise# 可选：重新抛出异常，让调用方感知并处理exceptExceptionase:print(f"❌ 操作文件出错：{e}")raisefinally:iffile_handle:file_handle.close()print(f"❌ 同步关闭文件：{file_path}")

扩展示例

演示contextmanager用于管理临时目录（创建→使用→自动删除），更贴近实际项目中的资源管理场景：

importosimportshutilfromcontextlibimportcontextmanager@contextmanagerdeftemporary_directory(dir_path:str="temp_dir"):"""同步临时目录管理器：自动创建→使用→删除临时目录"""# 1. 初始化：创建临时目录try:os.makedirs(dir_path,exist_ok=True)print(f"✅ 成功创建临时目录：{dir_path}")yielddir_path# 返回临时目录路径给with语句finally:# 2. 清理：删除临时目录及其内容ifos.path.exists(dir_path):shutil.rmtree(dir_path)print(f"❌ 成功删除临时目录：{dir_path}")# 使用临时目录deftest_temp_dir():withtemporary_directory()astemp_dir:# 在临时目录中创建文件temp_file=os.path.join(temp_dir,"test.txt")withopen(temp_file,"w")asf:f.write("临时文件内容")print(f"📄 临时文件已创建：{temp_file}，文件大小：{os.path.getsize(temp_file)}字节")if__name__=="__main__":test_temp_dir()