with 语句与上下文管理器

with open(...) as f: 是每个 Python 学习者的必经之路。但 with 的用途远不止”打开文件”——数据库连接、锁、临时切换目录、性能计时……几乎所有”需要成对进入/退出”的场景都能用它。这一篇讲清楚 with 背后的上下文管理器协议（__enter__ / __exit__），以及现代 Python 提供的 contextlib 工具箱。

一、with 是什么

with 语句保证”进入时执行一段代码、退出时（无论正常还是异常）执行另一段代码”。最经典的用途是自动关闭文件：

with open("data.txt", encoding="utf-8") as f:
    data = f.read()
# 到这里 f 已经自动关闭了

等价于：

f = open("data.txt", encoding="utf-8")
try:
    data = f.read()
finally:
    f.close()

with 更简洁、更不容易漏掉 close。

二、上下文管理器协议

任何实现了 __enter__ 和 __exit__ 方法的对象，都是上下文管理器：

class Timer:
    def __enter__(self):
        import time
        self.start = time.perf_counter()
        return self          # 会赋给 as 后的名字

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.end = time.perf_counter()
        print(f"耗时 {self.end - self.start:.4f}s")
        # 返回 True 会吞掉异常，返回 False/None 让异常继续传播

with Timer():
    sum(i * i for i in range(1_000_000))
# 耗时 0.0234s

关键点：

• __enter__(self) 返回的对象会被赋给 as 后的变量；
• __exit__(self, exc_type, exc_val, exc_tb)：即使 with 块内抛异常也会被调用；
• __exit__ 的返回值：True 表示”我处理了，别再抛”，其他值让异常继续传播。

三、exit 的三个参数

def __exit__(self, exc_type, exc_val, exc_tb):
    ...

• exc_type：异常类型（如 ValueError），没异常时为 None；
• exc_val：异常实例；
• exc_tb：traceback 对象。

利用它可以做异常翻译：

class SafeAPI:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ConnectionError:
            print(f"连接问题：{exc_val}，已重试")
            return True         # 吞掉
        return False            # 其他异常继续抛

别滥用”吞异常”——除非你真的知道自己在处理什么。

四、多个上下文管理器

with open("a.txt") as a, open("b.txt") as b:
    ...

# 括号跨行（Python 3.10+）
with (
    open("a.txt") as a,
    open("b.txt") as b,
    open("c.txt") as c,
):
    ...

打开顺序从左到右，关闭顺序从右到左（栈式）。

五、contextlib.contextmanager：用函数写上下文管理器

写类太麻烦，@contextmanager 让你用生成器函数实现：

from contextlib import contextmanager
import time

@contextmanager
def timer():
    start = time.perf_counter()
    try:
        yield              # yield 之前是 __enter__，之后是 __exit__
    finally:
        elapsed = time.perf_counter() - start
        print(f"耗时 {elapsed:.4f}s")


with timer():
    sum(i*i for i in range(1_000_000))

规则：

• yield 之前的代码 = __enter__ 里做的事；
• yield 的值会赋给 as 后的名字；
• yield 之后的代码 = __exit__ 里做的事；
• 用 try/finally 保证退出代码一定执行。

带值：

@contextmanager
def open_db(uri):
    conn = connect(uri)
    try:
        yield conn         # 会赋给 `with ... as conn`
    finally:
        conn.close()


with open_db("sqlite:///data.db") as conn:
    conn.execute("SELECT ...")

处理 with 块内的异常：

@contextmanager
def catch_and_log(logger):
    try:
        yield
    except Exception:
        logger.exception("with 块内出错")
        raise

六、常用的 contextlib 工具

suppress：忽略特定异常

from contextlib import suppress

with suppress(FileNotFoundError):
    Path("maybe.txt").unlink()

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

redirect_stdout / redirect_stderr

from contextlib import redirect_stdout
import io

buf = io.StringIO()
with redirect_stdout(buf):
    print("这段话不会显示在控制台")
    print("而是被捕获到 buf 里")

print("捕获到：", buf.getvalue())

nullcontext：占位

有时候某些代码路径需要”什么都不做”的上下文：

from contextlib import nullcontext

ctx = open("data.txt") if need_file else nullcontext()
with ctx as f:
    ...

ExitStack：动态数量的 with

不知道要打开多少个上下文时：

from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    # files 里所有文件会在退出时依次关闭

七、几个经典的上下文管理器场景

1. 临时切换工作目录

from contextlib import contextmanager
import os
from pathlib import Path

@contextmanager
def cd(path):
    old = Path.cwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old)


with cd("/tmp"):
    # 这段代码在 /tmp 里运行
    print(Path.cwd())
# 出来后回到原目录

2. 数据库事务

@contextmanager
def transaction(conn):
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    else:
        conn.commit()


with transaction(conn) as tx:
    tx.execute("...")
    tx.execute("...")

3. 加锁

import threading

lock = threading.Lock()

with lock:
    # 独占访问共享资源
    critical_section()

threading.Lock 已经实现了上下文管理器协议，直接用。

4. 环境变量临时替换

@contextmanager
def env_var(name, value):
    old = os.environ.get(name)
    os.environ[name] = value
    try:
        yield
    finally:
        if old is None:
            os.environ.pop(name, None)
        else:
            os.environ[name] = old


with env_var("DEBUG", "1"):
    run_test()
# 出来后 DEBUG 恢复原状

5. 计时（性能测试）

@contextmanager
def timer(label=""):
    start = time.perf_counter()
    yield
    print(f"{label} 用时 {time.perf_counter() - start:.4f}s")


with timer("解析"):
    parse_big_file()

八、异步上下文管理器 async with

Python 3.5+ 支持异步：

import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.get(url) as resp:
        data = await resp.text()

对应的协议是 __aenter__ 和 __aexit__。写异步版：

from contextlib import asynccontextmanager

@asynccontextmanager
async def open_ws(url):
    ws = await connect_ws(url)
    try:
        yield ws
    finally:
        await ws.close()


async with open_ws("wss://...") as ws:
    ...

九、类实现 vs 函数实现

场景	选择
只包一段简单逻辑，一次进/出	@contextmanager 函数
需要复杂状态、多个方法	类（__enter__/__exit__）
想支持 stack.enter_context(x)	类

大部分场景 @contextmanager 更简洁。

十、__enter__ 可以返回任意值

class Session:
    def __enter__(self):
        return {"user": "咖飞", "start": time.time()}

    def __exit__(self, *args):
        pass


with Session() as info:
    print(info["user"])

as 后面的名字接收 __enter__ 的返回值，不必是 self。

十一、一个完整例子：文件锁

# filelock.py
"""进程级文件锁（简化版，跨平台请用 filelock 库）。"""

import time
from pathlib import Path
from contextlib import contextmanager


@contextmanager
def file_lock(path: Path, timeout: float = 10):
    """基于目录存在与否的粗粒度锁。"""
    lock_dir = Path(f"{path}.lock")
    deadline = time.time() + timeout

    while True:
        try:
            lock_dir.mkdir()      # 目录不存在时创建，已存在则报错
            break
        except FileExistsError:
            if time.time() > deadline:
                raise TimeoutError(f"获取 {path} 锁超时")
            time.sleep(0.1)

    try:
        yield
    finally:
        try:
            lock_dir.rmdir()
        except OSError:
            pass


# 使用
if __name__ == "__main__":
    with file_lock(Path("shared.txt")):
        # 独占访问 shared.txt
        Path("shared.txt").write_text("safe write", encoding="utf-8")

注意：这只是演示，生产环境用 filelock 或 portalocker 这类库更可靠。

十二、常见陷阱

1. __enter__ 忘 return self：with X() as x 里 x 是 None。
2. @contextmanager 里没 try/finally：抛异常时清理代码跳过。
3. __exit__ 里再抛异常：会覆盖原异常。要么让原异常传，要么明确 raise ... from。
4. __exit__ 返回 True 吞异常：调用方莫名其妙，慎用。
5. with 里返回 / break：__exit__ 依然会执行（这是它的强项）。
6. 多个 with 时忽略关闭顺序：先开的最后关（栈式）。
7. 在异步代码用普通 with：应该用 async with。

十三、小结与延伸阅读

• with 保证”进入-退出”成对执行；
• 协议：__enter__ / __exit__；
• 函数式实现用 @contextmanager + 生成器；
• contextlib 提供 suppress、redirect_stdout、ExitStack、nullcontext；
• 场景：文件、锁、事务、切目录、临时环境、计时；
• 异步版：async with + __aenter__/__aexit__；
• 别乱吞异常。

延伸阅读：

• with 语句：https://docs.python.org/zh-cn/3/reference/compound_stmts.html#the-with-statement
• contextlib：https://docs.python.org/zh-cn/3/library/contextlib.html

到这里模块五（异常与文件处理）4 篇结束！ 下一篇进入 模块六：中级进阶，从 迭代器与可迭代对象 开始。