with语法与上下文管理

# 文件操作：离开 with 块时自动调用 f.close()
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()

# 同时管理多个资源（Python 3.10+ 可用括号跨行）
with (
    open("source.txt", "r", encoding="utf-8") as src,
    open("dest.txt", "w", encoding="utf-8") as dst,
):
    dst.write(src.read())

class DBConnection:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.conn = None

    def __enter__(self):
        print(f"连接数据库: {self.dsn}")
        self.conn = {"connected": True}   # 模拟连接
        return self.conn                  # 返回给 as 变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭连接（异常类型: {exc_type}）")
        self.conn = None
        return False   # 不吞掉异常

with DBConnection("postgres://localhost/app") as conn:
    print(f"使用连接: {conn}")
    # raise RuntimeError("查询失败")  # 即使发生异常，__exit__ 也会被调用

from contextlib import contextmanager

@contextmanager
def managed_resource(name: str):
    print(f"获取资源: {name}")    # 相当于 __enter__
    resource = {"name": name}
    try:
        yield resource             # yield 的值被 as 变量接收
    finally:
        print(f"释放资源: {name}") # 相当于 __exit__

with managed_resource("数据库连接") as r:
    print(f"使用: {r['name']}")

import time
from contextlib import contextmanager

@contextmanager
def timer(label: str = ""):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label} 耗时: {elapsed:.4f}s")

with timer("数据处理"):
    import time
    time.sleep(0.1)
    # 数据处理 耗时: 0.1001s

from contextlib import suppress
from pathlib import Path

# 等价于 try: ... except FileNotFoundError: pass
with suppress(FileNotFoundError):
    Path("nonexistent.txt").unlink()

# 忽略多种异常
with suppress(KeyError, IndexError):
    data = {}
    print(data["missing"])   # 静默忽略 KeyError

from contextlib import ExitStack

files = ["a.txt", "b.txt", "c.txt"]

with ExitStack() as stack:
    handles = [
        stack.enter_context(open(f, "w", encoding="utf-8"))
        for f in files
    ]
    for i, fh in enumerate(handles):
        fh.write(f"内容 {i}\n")
# 离开 with 块时，所有文件都会被自动关闭

from contextlib import nullcontext

def process(f=None):
    ctx = nullcontext(f) if f else open("default.txt", "r")
    with ctx as file:
        print(file.read())

from contextlib import contextmanager
from pathlib import Path

@contextmanager
def atomic_write(path: Path, encoding: str = "utf-8"):
    """先写到临时文件，成功后原子替换；失败则删除临时文件。"""
    tmp = path.with_suffix(path.suffix + ".tmp")
    try:
        with tmp.open("w", encoding=encoding) as f:
            yield f
        tmp.replace(path)   # 原子替换
    except Exception:
        if tmp.exists():
            tmp.unlink()
        raise

with atomic_write(Path("config.json")) as f:
    import json
    json.dump({"version": 2}, f, indent=2)