with 关键字
with 关键字
概述
Python 中的 with 语句用于上下文管理器（context manager）定义的方法包装块的执行，它允许将常见的 try...except...finally 使用模式封装起来以方便重用。with 语句确保在使用资源的代码完成运行时能够”清理”资源，即使抛出异常也是如此。
学习目标
通过本章学习，你将掌握：
- with 语句的基本语法和执行原理
- 上下文管理器的概念和实现
- 自定义上下文管理器的方法
- with 语句的实际应用场景
- 异常处理和资源管理的最佳实践
前置知识
- Python 基础语法
- 异常处理机制
- 类和对象的基本概念
- 文件操作基础
详细内容
基本语法
with EXPRESSION [as TARGET]:
SUITE
执行过程
with 语句的执行过程如下：
- 评估上下文表达式以获得上下文管理器
- 上下文管理器的
__enter__()被加载以备后用 - 上下文管理器的
__exit__()被加载以备后用 - 上下文管理器的
__enter__()方法被调用 - 如果 TARGET 包含在 with 语句中，则将
__enter__()的返回值赋值给它 - SUITE 被执行
- 上下文管理器的
__exit__()方法被调用
基本使用示例
文件操作
## 传统方式
file = open('example.txt', 'r')
try:
content = file.read()
print(content)
finally:
file.close()
## 使用 with 语句
with open('example.txt', 'r') as file:
content = file.read()
print(content)
## 文件会自动关闭,即使发生异常
多个资源管理
## 同时管理多个文件
with open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile:
data = infile.read()
outfile.write(data.upper())
## 两个文件都会自动关闭
自定义上下文管理器
基于类的上下文管理器
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()
# # 返回 False 表示不抑制异常
# # 返回 True 表示抑制异常
if exc_type is not None:
print(f"发生异常: {exc_type.__name__}: {exc_val}")
return False
## 使用自定义上下文管理器
with FileManager('test.txt', 'w') as f:
f.write('Hello, World!')
# # 如果这里发生异常,文件仍会被正确关闭
数据库连接管理器
class DatabaseConnection:
def __init__(self, host, port, database):
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 = f"connection_to_{self.database}"
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
print("关闭数据库连接")
if self.connection:
# # 这里应该是实际的连接关闭代码
self.connection = None
if exc_type is not None:
print(f"数据库操作异常: {exc_type.__name__}")
# # 可以在这里进行事务回滚
return False # 不抑制异常
else:
# # 可以在这里进行事务提交
print("数据库操作成功")
return False
## 使用数据库连接管理器
with DatabaseConnection('localhost', 5432, 'mydb') as conn:
print(f"使用连接: {conn}")
# # 执行数据库操作
使用 contextlib 模块
@contextmanager 装饰器
from contextlib import contextmanager
import time
@contextmanager
def timer():
start_time = time.time()
print("开始计时")
try:
yield start_time
finally:
end_time = time.time()
print(f"执行时间: {end_time - start_time:.2f} 秒")
## 使用计时器
with timer() as start:
time.sleep(1)
print("执行一些耗时操作")
临时目录管理
from contextlib import contextmanager
import tempfile
import shutil
import os
@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}")
## 临时目录和文件会自动删除
异常处理
异常信息获取
class ExceptionHandler:
def __enter__(self):
print("进入上下文")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("退出上下文")
if exc_type is not None:
print(f"异常类型: {exc_type}")
print(f"异常值: {exc_val}")
print(f"异常回溯: {exc_tb}")
# # 返回 True 抑制异常,返回 False 或 None 不抑制
return False
return False
## 测试异常处理
with ExceptionHandler():
print("正常执行")
raise ValueError("这是一个测试异常")
print("这行不会执行")
异常抑制示例
class IgnoreException:
def __init__(self, *exception_types):
self.exception_types = exception_types
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
if issubclass(exc_type, self.exception_types):
print(f"忽略异常: {exc_type.__name__}: {exc_val}")
return True # 抑制异常
return False
## 使用异常抑制
with IgnoreException(ValueError, TypeError):
print("开始执行")
raise ValueError("这个异常会被忽略")
print("这行不会执行")
print("程序继续执行")
实际应用场景
1. 线程锁管理
import threading
from contextlib import contextmanager
class ThreadSafeLock:
def __init__(self):
self._lock = threading.Lock()
def __enter__(self):
print("获取锁")
self._lock.acquire()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("释放锁")
self._lock.release()
return False
## 使用线程锁
lock = ThreadSafeLock()
shared_resource = 0
def worker():
global shared_resource
with lock:
# # 临界区代码
temp = shared_resource
temp += 1
shared_resource = temp
print(f"Worker updated resource to: {shared_resource}")
## 创建多个线程
threads = []
for i in range(3):
t = threading.Thread(target=worker)
threads.append(t)
t.start()
for t in threads:
t.join()
2. 配置管理
from contextlib import contextmanager
class ConfigManager:
def __init__(self):
self.config = {'debug': False, 'log_level': 'INFO'}
@contextmanager
def temporary_config(self, **kwargs):
# # 保存原始配置
original_config = self.config.copy()
# # 应用临时配置
self.config.update(kwargs)
print(f"应用临时配置: {kwargs}")
try:
yield self.config
finally:
# # 恢复原始配置
self.config = original_config
print("恢复原始配置")
## 使用配置管理器
config_manager = ConfigManager()
print(f"原始配置: {config_manager.config}")
with config_manager.temporary_config(debug=True, log_level='DEBUG'):
print(f"临时配置: {config_manager.config}")
# # 在这个块中使用调试配置
print(f"恢复后配置: {config_manager.config}")
3. 性能监控
from contextlib import contextmanager
import time
import psutil
import os
@contextmanager
def performance_monitor(operation_name):
# # 记录开始状态
start_time = time.time()
process = psutil.Process(os.getpid())
start_memory = process.memory_info().rss / 1024 / 1024 # MB
print(f"开始监控操作: {operation_name}")
print(f"初始内存使用: {start_memory:.2f} MB")
try:
yield
finally:
# # 记录结束状态
end_time = time.time()
end_memory = process.memory_info().rss / 1024 / 1024 # MB
execution_time = end_time - start_time
memory_diff = end_memory - start_memory
print(f"操作完成: {operation_name}")
print(f"执行时间: {execution_time:.2f} 秒")
print(f"内存变化: {memory_diff:+.2f} MB")
print(f"最终内存使用: {end_memory:.2f} MB")
## 使用性能监控
with performance_monitor("大数据处理"):
# # 模拟一些耗时和内存密集的操作
data = [i ** 2 for i in range(1000000)]
time.sleep(0.5)
result = sum(data)
print(f"计算结果: {result}")
4. HTTP 请求会话管理
from contextlib import contextmanager
import requests
@contextmanager
def http_session(base_url, headers=None, timeout=30):
session = requests.Session()
if headers:
session.headers.update(headers)
# # 设置基础 URL
session.base_url = base_url
print(f"创建 HTTP 会话: {base_url}")
try:
yield session
finally:
session.close()
print("关闭 HTTP 会话")
## 使用 HTTP 会话管理
headers = {
'User-Agent': 'MyApp/1.0',
'Accept': 'application/json'
}
with http_session('https://api.example.com', headers=headers) as session:
# # 在这个块中使用会话进行多个请求
# # response1 = session.get('/users')
# # response2 = session.post('/data', json={'key': 'value'})
print("执行 HTTP 请求...")
常见陷阱与最佳实践
1. exit 方法的返回值
class ExampleContext:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
print(f"捕获到异常: {exc_val}")
# # 返回 True 会抑制异常
# # 返回 False 或 None 会传播异常
return True # 抑制异常
return False
## 异常被抑制,程序继续执行
with ExampleContext():
raise ValueError("测试异常")
print("程序继续执行") # 这行会被执行
2. 嵌套上下文管理器
## 方法 1:嵌套 with 语句
with open('file1.txt', 'r') as f1:
with open('file2.txt', 'w') as f2:
data = f1.read()
f2.write(data.upper())
## 方法 2:使用逗号分隔(推荐)
with open('file1.txt', 'r') as f1, open('file2.txt', 'w') as f2:
data = f1.read()
f2.write(data.upper())
## 方法 3:使用 ExitStack(复杂场景)
from contextlib import ExitStack
with ExitStack() as stack:
f1 = stack.enter_context(open('file1.txt', 'r'))
f2 = stack.enter_context(open('file2.txt', 'w'))
data = f1.read()
f2.write(data.upper())
3. 资源清理的重要性
## 错误示例:没有正确清理资源
class BadResource:
def __init__(self, name):
self.name = name
self.resource = f"resource_{name}"
def __enter__(self):
print(f"获取资源: {self.resource}")
return self.resource
def __exit__(self, exc_type, exc_val, exc_tb):
# # 忘记清理资源
pass
## 正确示例:确保资源清理
class GoodResource:
def __init__(self, name):
self.name = name
self.resource = None
def __enter__(self):
self.resource = f"resource_{self.name}"
print(f"获取资源: {self.resource}")
return self.resource
def __exit__(self, exc_type, exc_val, exc_tb):
if self.resource:
print(f"清理资源: {self.resource}")
self.resource = None
return False
相关函数与模块
内置函数
open()- 文件操作的内置上下文管理器iter()- 某些迭代器也支持上下文管理协议
标准库模块
contextlib- 上下文管理工具@contextmanager- 装饰器，简化上下文管理器创建ExitStack- 动态管理多个上下文管理器closing()- 为有 close() 方法的对象提供上下文管理suppress()- 抑制指定异常
threading- 线程锁等同步原语都支持上下文管理multiprocessing- 进程锁等也支持上下文管理
第三方库
requests- HTTP 会话管理sqlite3- 数据库连接和事务管理psycopg2- PostgreSQL 数据库连接管理
扩展阅读
相关标签
Python 上下文管理器 资源管理 异常处理 with 语句 contextlib 文件操作 内存管理
讨论与反馈
欢迎在下方留言讨论，分享你的学习心得或提出问题。评论基于GitHub Issues，需要GitHub账号。