Python 中的 with open：上下文管理器深入解析

with open('data.txt', 'r') as f:
    content = f.read()

# ❌ 容易出错的写法
f = open('data.txt', 'r')
content = f.read()
# 如果 read() 抛出异常，close() 永远不会被调用！
f.close()

# ⚠️ 勉强安全的写法
f = open('data.txt', 'r')
try:
    content = f.read()
finally:
    f.close()  # 即使异常也会关闭

# ✅ 使用 with 语句
with open('data.txt', 'r') as f:
    content = f.read()
# 离开 with 块时，文件自动关闭
# 即使 read() 抛出异常，也会确保关闭

with open(
    file='data.txt',        # 文件路径
    mode='r',               # 打开模式
    buffering=-1,           # 缓冲策略
    encoding='utf-8',       # 编码
    errors='strict',        # 编码错误处理
    newline=None,           # 换行符处理
    closefd=True,           # 是否关闭底层文件描述符
    opener=None             # 自定义打开器
) as f:
    data = f.read()

# 方法1：嵌套 with
with open('input.txt', 'r') as fin:
    with open('output.txt', 'w') as fout:
        fout.write(fin.read().upper())

# 方法2：单行多个（Python 3.1+）
with open('input.txt', 'r') as fin, open('output.txt', 'w') as fout:
    fout.write(fin.read().upper())

# 方法3：使用 contextlib.ExitStack（动态数量）
from contextlib import ExitStack

filenames = ['a.txt', 'b.txt', 'c.txt']
with ExitStack() as stack:
    files = [stack.enter_context(open(f, 'r')) for f in filenames]
    # 所有文件在这里都打开了
    for f in files:
        print(f.readline())
# 离开 with 时，所有文件按 LIFO 顺序关闭

# Python 解释器实际执行的等价伪代码
_manager = open('data.txt', 'r')      # 创建管理器对象
f = _manager.__enter__()              # 进入上下文
try:
    content = f.read()                # 执行 with 块
except Exception as e:
    if not _manager.__exit__(         # 发生异常
        type(e), e, e.__traceback__
    ):
        raise                          # __exit__ 返回 False，继续抛出
else:
    _manager.__exit__(None, None, None)  # 正常退出

class DatabaseConnection:
    """模拟数据库连接"""

    def __init__(self, db_url):
        self.db_url = db_url
        self.connection = None

    def __enter__(self):
        print(f"正在连接数据库: {self.db_url}")
        self.connection = {"status": "connected", "url": self.db_url}
        return self.connection  # 这个返回值赋给 as 后的变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("正在关闭数据库连接...")
        self.connection["status"] = "disconnected"

        # 处理特定异常
        if exc_type is not None:
            print(f"发生异常: {exc_type.__name__}: {exc_val}")
            # 返回 False 让异常继续传播
            return False

# 使用
with DatabaseConnection("postgresql://localhost/mydb") as conn:
    print(f"连接状态: {conn['status']}")
    # 执行业务逻辑...
# 输出:
# 正在连接数据库: postgresql://localhost/mydb
# 连接状态: connected
# 正在关闭数据库连接...

from contextlib import contextmanager

@contextmanager
def database_connection(db_url):
    """使用生成器实现的上下文管理器"""
    print(f"正在连接数据库: {db_url}")
    conn = {"status": "connected", "url": db_url}

    try:
        yield conn  # 这里将 conn 交给 with 块
    except Exception as e:
        print(f"捕获到异常: {e}")
        # 这里可以选择重新抛出或抑制
        raise
    finally:
        print("正在关闭数据库连接...")
        conn["status"] = "disconnected"

# 使用方式完全相同
with database_connection("postgresql://localhost/mydb") as conn:
    print(f"连接状态: {conn['status']}")

import tempfile
import shutil
import os
from contextlib import contextmanager

@contextmanager
def temporary_directory():
    """创建临时目录，使用完毕自动清理"""
    tmp_dir = tempfile.mkdtemp()
    print(f"创建临时目录: {tmp_dir}")

    try:
        yield tmp_dir
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)
        print(f"临时目录已删除: {tmp_dir}")

# 使用
with temporary_directory() as tmp:
    # 在临时目录中做一些操作
    with open(os.path.join(tmp, 'temp.txt'), 'w') as f:
        f.write('临时数据')
    print(f"临时目录内容: {os.listdir(tmp)}")
# 退出时自动删除

import time
from contextlib import contextmanager

@contextmanager
def timer(description="操作"):
    """测量代码块执行时间"""
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"{description} 耗时: {elapsed:.4f} 秒")

# 使用
with timer("大数据排序"):
    data = list(range(10_000_000))
    data.sort(reverse=True)
# 输出: 大数据排序 耗时: 0.4523 秒

import threading
from contextlib import contextmanager

lock = threading.Lock()

@contextmanager
def locked():
    """自动获取和释放锁"""
    lock.acquire()
    try:
        yield
    finally:
        lock.release()

# 使用
counter = 0

def increment():
    global counter
    with locked():
        counter += 1

import sqlite3
from contextlib import contextmanager

@contextmanager
def transaction(conn):
    """自动提交或回滚事务"""
    try:
        yield conn
        conn.commit()
        print("事务已提交")
    except Exception as e:
        conn.rollback()
        print(f"事务已回滚: {e}")
        raise

# 使用
conn = sqlite3.connect('example.db')
try:
    with transaction(conn):
        conn.execute("INSERT INTO users (name) VALUES ('Alice')")
        conn.execute("INSERT INTO users (name) VALUES ('Bob')")
        # 如果这里抛出异常，两条 INSERT 都会回滚
        raise ValueError("模拟错误")
except ValueError:
    print("已正确处理异常，事务已回滚")

import io
from contextlib import redirect_stdout

# 捕获 print 输出
buf = io.StringIO()
with redirect_stdout(buf):
    print("这条输出被重定向了")
    print("不会显示在终端")

print(f"捕获的内容: {buf.getvalue()}")

from contextlib import suppress

# 忽略指定类型的异常
with suppress(FileNotFoundError):
    os.remove('may_not_exist.txt')
# 等价于:
# try:
#     os.remove('may_not_exist.txt')
# except FileNotFoundError:
#     pass

# Python 3.11+
from contextlib import chdir
import os

print(f"当前目录: {os.getcwd()}")
with chdir('/tmp'):
    print(f"在 with 块内: {os.getcwd()}")
print(f"恢复: {os.getcwd()}")

# ❌ 错误：文件已经关闭
with open('data.txt', 'r') as f:
    data = f.readlines()

for line in data:  # ✅ 可以，data 是内存中的数据
    print(line)

# 但：
with open('data.txt', 'r') as f:
    pass

line = f.readline()  # ❌ ValueError: I/O operation on closed file

class DangerousManager:
    def __enter__(self):
        return self

    def __exit__(self, *args):
        return True  # ❌ 吞掉所有异常，包括 SystemExit！

# 这会导致程序无法正常退出

@contextmanager
def bad_manager():
    resource = acquire_resource()
    yield resource
    # ⚠️ 如果 with 块中抛出异常，这里不会执行！
    # 始终把清理代码放在 try/finally 中
    release_resource(resource)

@contextmanager
def good_manager():
    resource = acquire_resource()
    try:
        yield resource
    finally:
        # ✅ 无论如何都会执行
        release_resource(resource)