Python文件操作之with语句上下文管理器自动关闭文件

# ❌ 传统方式——需要手动关闭
f = open("test.txt", "w", encoding="utf-8")
try:
    f.write("Hello\n")
    # ... 如果这里发生异常 ...
    f.write("World\n")
finally:
    f.close()  # 即使异常也要确保关闭

# ✅ with语句——自动关闭
with open("test.txt", "w", encoding="utf-8") as f:
    f.write("Hello\n")
    # 即使这里发生异常，文件也会自动关闭
    f.write("World\n")
# 出了with块，文件自动关闭

# 验证文件已关闭
# f.write("test")  # ValueError: I/O operation on closed file.

# 文件对象实现了上下文管理器协议
# with open(...) as f 等价于：

f = open("test.txt", "w", encoding="utf-8")
try:
    # with块中的代码
    f.write("Hello\n")
finally:
    f.close()  # 无论是否异常，都执行

# 💡 with语句背后：
# 1. open()返回的文件对象实现了__enter__和__exit__
# 2. __enter__返回文件对象本身（赋值给as后的变量）
# 3. 退出with块时，__exit__自动调用close()
# 4. 即使with块内发生异常，__exit__也会执行

# with语句最核心的价值——异常安全

def demonstrate_with_safety():
    """演示with的异常安全"""
    try:
        with open("test.txt", "w", encoding="utf-8") as f:
            f.write("开始写入\n")
            # 模拟错误
            raise RuntimeError("模拟程序崩溃")
            f.write("这行永远不会执行\n")
    except RuntimeError as e:
        print(f"捕获异常: {e}")

    # 文件已经被正确关闭了
    with open("test.txt", "r", encoding="utf-8") as f:
        content = f.read()
        print(f"文件内容: {repr(content)}")
        # '开始写入\n' —— 异常前的写入被保存了

demonstrate_with_safety()

# 同时打开多个文件——用逗号分隔

with open("source.txt", "r", encoding="utf-8") as src, \
     open("destination.txt", "w", encoding="utf-8") as dst:
    for line in src:
        dst.write(line.upper())

# 或者使用括号（Python会自动连接）
with (
    open("a.txt", "r", encoding="utf-8") as fa,
    open("b.txt", "r", encoding="utf-8") as fb,
    open("merged.txt", "w", encoding="utf-8") as fout,
):
    fout.write(fa.read())
    fout.write(fb.read())

# 💡 所有文件在with块结束时都会自动关闭
# 如果打开多个文件时发生异常（如文件不存在），已成功打开的文件也会被关闭

# 嵌套with——每个文件独立的生命周期

def process_with_temp(source_path):
    """处理文件——使用临时文件"""
    import os

    temp_path = source_path + ".tmp"

    # 外层with：打开源文件
    with open(source_path, "r", encoding="utf-8") as src:
        # 内层with：打开临时文件
        with open(temp_path, "w", encoding="utf-8") as tmp:
            for line in src:
                processed = line.strip().upper()
                tmp.write(processed + "\n")

    # 两个文件都已关闭——可以安全替换
    os.replace(temp_path, source_path)

# 支持with的对象可以用于条件判断

def maybe_open_file(filepath=None):
    """可能打开文件也可能不打开"""
    if filepath:
        # 用with管理
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    else:
        # contextlib.nullcontext——一个"什么都不做"的上下文管理器
        from contextlib import nullcontext
        with nullcontext():
            return "没有文件"

# nullcontext在需要"可选的上下文管理器"时非常有用

import json
import os

class ConfigManager:
    """配置文件管理器——用with保证安全"""

    def __init__(self, filepath):
        self.filepath = filepath

    def load(self):
        """加载配置"""
        if not os.path.exists(self.filepath):
            return {}

        with open(self.filepath, "r", encoding="utf-8") as f:
            return json.load(f)

    def save(self, config):
        """安全保存配置——先写临时文件"""
        temp_path = self.filepath + ".tmp"

        with open(temp_path, "w", encoding="utf-8") as f:
            json.dump(config, f, ensure_ascii=False, indent=2)
            f.flush()
            os.fsync(f.fileno())

        # with块结束，文件已关闭——安全替换
        os.replace(temp_path, self.filepath)

# 使用with实现文件锁——防止多进程同时操作同一文件

import os
import time

class FileLock:
    """文件锁——通过创建锁文件来防止并发"""

    def __init__(self, lock_path, timeout=10):
        self.lock_path = lock_path
        self.timeout = timeout
        self.locked = False

    def __enter__(self):
        """尝试获取锁"""
        start = time.time()
        while True:
            try:
                # 独占创建锁文件——如果已存在则报错
                with open(self.lock_path, "x") as f:
                    f.write(str(os.getpid()))
                self.locked = True
                return self
            except FileExistsError:
                if time.time() - start > self.timeout:
                    raise TimeoutError("获取锁超时")
                time.sleep(0.1)

    def __exit__(self, *args):
        """释放锁"""
        if self.locked:
            os.remove(self.lock_path)
            self.locked = False

# 使用
# with FileLock("/tmp/my_app.lock"):
#     # 在锁保护下执行操作
#     do_critical_operation()