Python上下文管理器 - with语句

# 基本语法
with expression as variable:
    # 代码块
    do_something(variable)

# expression 返回一个上下文管理器对象
# as variable 是可选的，用于将上下文管理器的结果绑定到变量

# 传统写法（不推荐）
f = open("test.txt", "r", encoding="utf-8")
try:
    content = f.read()
    print(content)
finally:
    f.close()

# 使用with语句（推荐）
with open("test.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)
# with语句会自动调用f.close()，无需手动关闭

# 使用with语句写入文件
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.write("这是with语句的写入示例\n")

# 读取刚才写入的内容
with open("output.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())

# with语句在发生异常时也会自动关闭文件
with open("data.txt", "w", encoding="utf-8") as f:
    f.write("第一行数据\n")
    f.write("第二行数据\n")
    raise ValueError("模拟异常")  # 即使抛出异常，文件也会被正确关闭

print("文件已自动关闭")

class MyContextManager:
    """自定义上下文管理器示例"""
    
    def __init__(self, name):
        self.name = name
    
    def __enter__(self):
        print(f"进入上下文: {self.name}")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"退出上下文: {self.name}")
        # 返回True表示异常已处理，不会继续向外传播
        # 返回False或None表示异常需要继续向外传播
        if exc_type is not None:
            print(f"捕获到异常: {exc_type.__name__}: {exc_val}")
        return False

# 使用自定义上下文管理器
with MyContextManager("测试上下文") as manager:
    print(f"在上下文中操作: {manager.name}")

print("---" * 20)

# 测试异常处理
with MyContextManager("异常测试") as manager:
    print("准备抛出异常...")
    raise RuntimeError("模拟错误")

class DatabaseConnection:
    """模拟数据库连接管理器"""
    
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None
    
    def __enter__(self):
        print(f"正在连接到数据库: {self.db_name}")
        # 模拟建立连接
        self.connection = {"db": self.db_name, "status": "connected"}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"正在关闭数据库连接: {self.db_name}")
        self.connection["status"] = "closed"
        # 如果发生异常，可以选择回滚事务
        if exc_type is not None:
            print("事务回滚中...")
        return False

# 使用数据库连接管理器
with DatabaseConnection("my_database") as conn:
    print(f"连接状态: {conn['status']}")
    # 执行数据库操作...
    print("执行查询操作...")

from contextlib import contextmanager

@contextmanager
def file_manager(filename, mode):
    """使用contextmanager装饰器创建文件管理器"""
    print(f"打开文件: {filename}")
    f = open(filename, mode, encoding="utf-8")
    try:
        yield f  # yield之前的代码相当于__enter__
    finally:
        print(f"关闭文件: {filename}")
        f.close()  # yield之后的代码相当于__exit__

# 使用装饰器创建的文件管理器
with file_manager("test.txt", "w") as f:
    f.write("使用contextmanager创建的上下文管理器\n")

from contextlib import contextmanager
import time

@contextmanager
def timer(description="操作"):
    """使用contextmanager创建计时器"""
    start = time.time()
    try:
        yield
    finally:
        elapsed = time.time() - start
        print(f"{description}耗时: {elapsed:.4f}秒")

# 使用计时器
with timer("数据处理"):
    # 模拟耗时操作
    time.sleep(0.5)
    total = sum(range(1000000))
    print(f"计算结果: {total}")

# 传统的嵌套写法
with open("input.txt", "r", encoding="utf-8") as infile:
    with open("output.txt", "w", encoding="utf-8") as outfile:
        content = infile.read()
        outfile.write(content.upper())
        print("文件复制完成")

# 合并写法（更简洁）
with open("input.txt", "r", encoding="utf-8") as infile, \
     open("output.txt", "w", encoding="utf-8") as outfile:
    content = infile.read()
    outfile.write(content.upper())
    print("文件复制完成")