字数 2184，阅读大约需 11 分钟
Python 中的上下文管理器是什么？
参考资料
Python 中的上下文管理器是什么？[1]
深入理解 Python 中的上下文管理器 - 王一白 - 博客园[2]
在Python编程中，我们经常需要处理文件、数据库连接、网络连接等资源。这些资源在使用完毕后必须正确释放，否则可能导致内存泄漏、文件句柄耗尽等问题。Python的上下文管理器（Context Manager）为我们提供了一种优雅、安全的方式来管理这些资源。
什么是上下文管理器？
上下文管理器是一种Python对象，它定义了在with语句中使用时的运行时上下文。它确保无论代码块如何退出（正常退出还是异常退出），都能执行必要的清理操作。
1. 自动化资源管理：确保资源的正确获取和释放
2. 异常安全：即使发生异常也能正确清理
3. 代码简洁：避免重复的try-finally模式
4. 可读性强：明确表达资源的生命周期
核心概念
上下文管理器必须实现两个特殊方法：
•
__enter__()：进入上下文时调用•
__exit__(exc_type, exc_val, exc_tb)：退出上下文时调用
为什么需要上下文管理器？
让我们先看看不使用上下文管理器可能遇到的问题：
# 不安全的文件操作
def read_file_unsafe(filename):
file = open(filename, 'r')
data = file.read()
# 如果这里发生异常，文件将不会被关闭
process_data(data)
file.close() # 可能永远不会执行
return data使用上下文管理器的安全版本：
# 安全的文件操作
def read_file_safe(filename):
with open(filename, 'r') as file:
data = file.read()
process_data(data)
# 无论是否发生异常，文件都会被正确关闭
return datawith语句的工作原理
with语句的执行流程如下：
1. 计算上下文表达式，获得上下文管理器对象
2. 调用上下文管理器的
__enter__()方法3. 将
__enter__()的返回值赋给as后的变量（如果有）4. 执行with语句体
5. 调用上下文管理器的
__exit__()方法进行清理
# with语句的等价形式
manager = open('file.txt', 'r')
try:
file = manager.__enter__()
# 执行with语句体
data = file.read()
finally:
manager.__exit__(None, None, None)内置的上下文管理器
1. 文件操作
# 读取文件
with open('data.txt', 'r', encoding='utf-8') as f:
content = f.read()
print(content)
# 文件自动关闭
# 写入文件
with open('output.txt', 'w', encoding='utf-8') as f:
f.write('Hello, World!')
# 文件自动关闭并保存2. 线程锁
import threading
lock = threading.Lock()
# 使用上下文管理器自动获取和释放锁
with lock:
# 临界区代码
shared_resource += 1
# 锁自动释放3. 数据库连接
import sqlite3
# 数据库连接自动管理
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')
results = cursor.fetchall()
# 连接自动提交和关闭自定义上下文管理器
方法1：类实现
class FileManager:
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
print(f"打开文件: {self.filename}")
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
print(f"关闭文件: {self.filename}")
if self.file:
self.file.close()
# 处理异常
if exc_type is not None:
print(f"发生异常: {exc_type.__name__}: {exc_val}")
# 返回False表示不抑制异常，返回True表示抑制异常
return False
# 使用自定义上下文管理器
with FileManager('test.txt', 'w') as f:
f.write('Hello Context Manager!')方法2：使用contextlib.contextmanager装饰器
from contextlib import contextmanager
@contextmanager
def database_connection(db_name):
print(f"连接到数据库: {db_name}")
conn = f"connection_to_{db_name}" # 模拟数据库连接
try:
yield conn # yield之前的代码相当于__enter__
finally:
print(f"关闭数据库连接: {db_name}") # yield之后的代码相当于__exit__
# 使用
with database_connection('mydb') as conn:
print(f"使用连接: {conn}")
# 可能的数据库操作高级用法示例
1. 计时器上下文管理器
import time
from contextlib import contextmanager
@contextmanager
def timer():
start_time = time.time()
print("开始计时...")
try:
yield
finally:
end_time = time.time()
print(f"执行时间: {end_time - start_time:.4f}秒")
# 使用计时器
with timer():
time.sleep(1)
print("执行一些耗时操作")2. 临时目录管理器
import os
import tempfile
import shutil
from contextlib import contextmanager
@contextmanager
def temporary_directory():
temp_dir = tempfile.mkdtemp()
print(f"创建临时目录: {temp_dir}")
try:
yield temp_dir
finally:
print(f"删除临时目录: {temp_dir}")
shutil.rmtree(temp_dir)
# 使用临时目录
with temporary_directory() as temp_dir:
# 在临时目录中工作
file_path = os.path.join(temp_dir, 'temp_file.txt')
with open(file_path, 'w') as f:
f.write('临时数据')
print(f"临时文件创建: {file_path}")
# 临时目录及其内容会被自动删除3. 异常处理和日志记录
import logging
from contextlib import contextmanager
@contextmanager
def logged_operation(operation_name):
logging.info(f"开始操作: {operation_name}")
start_time = time.time()
try:
yield
logging.info(f"操作成功完成: {operation_name}")
except Exception as e:
logging.error(f"操作失败: {operation_name}, 错误: {e}")
raise # 重新抛出异常
finally:
duration = time.time() - start_time
logging.info(f"操作耗时: {duration:.4f}秒")
# 使用
with logged_operation("数据处理"):
# 执行可能出错的操作
data = [1, 2, 3, 4, 5]
result = sum(data)
print(f"结果: {result}")多个上下文管理器
嵌套使用
# 传统嵌套方式
with open('input.txt', 'r') as infile:
with open('output.txt', 'w') as outfile:
data = infile.read()
outfile.write(data.upper())使用contextlib.ExitStack
from contextlib import ExitStack
# 更优雅的多重上下文管理
with ExitStack() as stack:
infile = stack.enter_context(open('input.txt', 'r'))
outfile = stack.enter_context(open('output.txt', 'w'))
data = infile.read()
outfile.write(data.upper())Python 3.10+的新语法
# Python 3.10及以上版本支持
with (
open('input.txt', 'r') as infile,
open('output.txt', 'w') as outfile
):
data = infile.read()
outfile.write(data.upper())异步上下文管理器
对于异步操作，Python提供了异步上下文管理器：
import asyncio
import aiohttp
class AsyncHTTPClient:
def __init__(self, base_url):
self.base_url = base_url
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def get(self, endpoint):
async with self.session.get(f"{self.base_url}/{endpoint}") as response:
return await response.text()
# 使用异步上下文管理器
async def main():
async with AsyncHTTPClient('https://api.example.com') as client:
data = await client.get('users')
print(data)
# asyncio.run(main())实际应用场景
1. 数据库事务管理
from contextlib import contextmanager
import sqlite3
@contextmanager
def transaction(connection):
try:
yield connection
connection.commit()
print("事务提交成功")
except Exception as e:
connection.rollback()
print(f"事务回滚: {e}")
raise
# 使用事务管理器
conn = sqlite3.connect('example.db')
with transaction(conn):
cursor = conn.cursor()
cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
cursor.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
# 如果任何操作失败，整个事务都会回滚2. 配置环境管理
import os
from contextlib import contextmanager
@contextmanager
def environment_variables(**kwargs):
# 保存原始环境变量
original_env = {}
for key, value in kwargs.items():
original_env[key] = os.environ.get(key)
os.environ[key] = value
try:
yield
finally:
# 恢复原始环境变量
for key, original_value in original_env.items():
if original_value is None:
os.environ.pop(key, None)
else:
os.environ[key] = original_value
# 使用环境变量管理器
with environment_variables(DEBUG='True', LOG_LEVEL='INFO'):
# 在这个上下文中，环境变量被临时修改
print(f"DEBUG: {os.environ.get('DEBUG')}")
print(f"LOG_LEVEL: {os.environ.get('LOG_LEVEL')}")
# 环境变量自动恢复到原始状态最佳实践
1. 确保异常安全
@contextmanager
def safe_resource():
resource = acquire_resource()
try:
yield resource
except Exception as e:
# 处理特定异常
handle_exception(e)
raise
finally:
# 无论如何都要清理资源
cleanup_resource(resource)2. 提供有用的错误信息
class ConfigManager:
def __init__(self, config_file):
self.config_file = config_file
self.config = None
def __enter__(self):
try:
with open(self.config_file, 'r') as f:
self.config = json.load(f)
return self.config
except FileNotFoundError:
raise FileNotFoundError(f"配置文件不存在: {self.config_file}")
except json.JSONDecodeError as e:
raise ValueError(f"配置文件格式错误: {e}")
def __exit__(self, exc_type, exc_val, exc_tb):
# 清理操作
self.config = None3. 支持参数化
@contextmanager
def database_transaction(connection, isolation_level='READ_COMMITTED'):
old_isolation = connection.isolation_level
connection.set_isolation_level(isolation_level)
transaction = connection.begin()
try:
yield connection
transaction.commit()
except Exception:
transaction.rollback()
raise
finally:
connection.set_isolation_level(old_isolation)常见陷阱和注意事项
1. 不要在__enter__中抛出异常后忘记清理
# 错误示例
class BadManager:
def __enter__(self):
self.resource = acquire_expensive_resource()
if not validate_resource(self.resource):
# 直接抛出异常，资源没有被清理
raise ValueError("资源无效")
return self.resource
def __exit__(self, exc_type, exc_val, exc_tb):
release_resource(self.resource)
# 正确示例
class GoodManager:
def __enter__(self):
self.resource = acquire_expensive_resource()
try:
if not validate_resource(self.resource):
raise ValueError("资源无效")
return self.resource
except:
# 如果验证失败，清理资源
release_resource(self.resource)
raise
def __exit__(self, exc_type, exc_val, exc_tb):
release_resource(self.resource)2. 正确处理__exit__的返回值
@contextmanager
def suppress_errors(*exceptions):
try:
yield
except exceptions as e:
print(f"抑制异常: {e}")
# 返回True表示抑制异常
return True
# 隐式返回None (False)，不抑制其他异常
# 使用
with suppress_errors(ValueError, TypeError):
int("not a number") # 这个异常会被抑制
print("这行代码会执行")总结
上下文管理器是Python中一个强大而优雅的特性，它帮助我们：
1. 自动化资源管理：确保资源的正确获取和释放
2. 异常安全：即使发生异常也能正确清理
3. 代码简洁：避免重复的try-finally模式
4. 可读性强：明确表达资源的生命周期
掌握上下文管理器的使用，不仅能让你的代码更加健壮和优雅，也是成为Python高手的必经之路。在日常开发中，多思考哪些场景可以使用上下文管理器来改善代码质量，你会发现它们无处不在的价值。
引用链接
[1] Python 中的上下文管理器是什么？: https://www.freecodecamp.org/news/context-managers-in-python/[2] 深入理解 Python 中的上下文管理器 - 王一白 - 博客园: https://www.cnblogs.com/wongbingming/p/10519553.html
评论区