把"分配→使用→必释放"焊进语法: __enter__ 进, __exit__ 必被调用(异常也执行) — 嵌套即栈, 逆序退出
with 是把"分配 → 使用 → 必释放"焊死在语法里的结构: 进入块时调 __enter__ 拿资源, 退出时无论 body 是正常走完、return 还是抛异常, __exit__ 都保证被调用 —— 相当于解释器替你写好了 try/finally。异常发生时, 它的类型/值/回溯以三个参数交给 __exit__, 返回 True 才会吞掉异常, 否则继续向上传播。多个 with 嵌套就是一个资源栈: 内层先退、外层后退; ExitStack 更进一步, 让运行时才知道个数的 N 个资源也能统一收尾。写对 __enter__/__exit__ 的位置与返回值, 是所有事务、锁、连接池模板的共同地基。
mgr = expr; v = mgr.__enter__(), try 执行 body; 异常时 mgr.__exit__(type, val, tb) 返回假值则重新 raise, 正常结束时调 __exit__(None, None, None)。 mgr = open("cfg.yaml") # 关键: with 只是这段的语法糖 v = mgr.__enter__() # as v 绑定它的返回值 try: process(v) # body: 正常/return/异常三条路 finally: mgr.__exit__(exc_type, exc_val, tb) # 异常也必被调用
__init__, 保证"进块才占用", 对象本身可以安全传递与复用。 class Conn: def __init__(self, dsn): # 只存参数, 不占资源 self.dsn = dsn def __enter__(self): self.c = connect(self.dsn) # 关键: 进块才真正拿连接 return self.c
exc_type(异常类)/exc_val(异常实例)/exc_tb(traceback 对象); 无异常退出时三个全是 None — 用它可以区分"正常收尾"与"带病收尾"。 def __exit__(self, exc_type, exc_val, tb): if exc_type is None: # 关键: 三个 None = 正常收尾 self.commit() else: # 非 None = 带病收尾 self.rollback() # exc_val 就是异常实例
def __exit__(self, t, v, tb): self.close() # 不写 return → 返回 None # → None/False: 异常继续上抛(清理型默认, 正确) # → return True: body 的异常被吞, 上层无感知(慎用)
yield 之前的代码≈__enter__, 之后的(通常放 finally)≈__exit__; body 抛的异常会在 yield 表达式处重新抛出。 @contextmanager def cd(path): old = os.getcwd() # yield 前 ≈ __enter__ os.chdir(path) try: yield # 关键: body 异常在此抛回生成器 finally: os.chdir(old) # ≈ __exit__, 异常也执行
enter_context(cm) 注册 CM, callback(fn) 注册裸回调, 退出时严格 LIFO 全部执行 — 分支/循环中获取的资源靠它收口。 with ExitStack() as st: f1 = st.enter_context(open("a.log")) # 先注册 → 后关 f2 = st.enter_context(open("b.log")) # 后注册 → 先关 st.callback(print, "bye") # 裸回调同样进栈 # 关键: 退出顺序 bye → f2 → f1, 严格 LIFO
suppress(E) 精准吞一种异常; closing(obj) 用完调 close; nullcontext 空操作占位; redirect_stdout 临时换标准输出。 with suppress(FileNotFoundError): os.unlink(sock_path) # 只吞这一种, 其余照抛 with closing(open_url()) as r: # 出块自动 r.close() data = r.read() with nullcontext() as x: # 空操作占位, CM 可选时对齐 pass
__aenter__/__aexit__ 两个协程方法, await 进出; contextlib.asynccontextmanager 是生成器写法的异步版, 配 AsyncExitStack 管动态资源。 class Conn: async def __aenter__(self): # await 进入 self.c = await connect(dsn) return self.c async def __aexit__(self, t, v, tb): # await 退出 await self.c.close() # 返回值语义同 __exit__
with A(): with B(): 的退出顺序是 B 先 A 后 — 与获取顺序相反, 和函数调用栈同构, 所以"后获取的依赖先获取的"不会倒挂。 with conn: # A 先进 with cursor: # B 后进 write_rows() # 关键: 退出顺序 cursor 先、conn 后 — LIFO 逆序
cm = Conn("db") with cm: ... # 第一次 __enter__ 建连接 with cm: ... # 错: 复用实例, 状态被覆盖 # 关键: CM 默认一次性 — 惯用法 with Conn("db"): # 复用须显式设计成可重入(深度计数)
@contextmanager def leak(): f = open("a.log") yield f # 关键: body 异常在这行抛出 f.close() # → 异常时不执行, 泄漏! # 修法: close 放进 finally
with open("a.txt") as f: # ← 等价于 ↓ data = f.read() f = open("a.txt") try: data = f.read() finally: f.close() # 排查时心里过一遍展开式
业务代码手写 commit/except rollback 三件套, 十个人写出五种遗漏。收进一个 @contextmanager 后, 事务纪律成了语法:
from contextlib import contextmanager @contextmanager def tx(conn): # 一个 with = 一个事务边界 conn.autocommit(False) # 为什么: 关自动提交, 失败可整体回滚 try: yield conn.cursor() conn.commit() # body 正常走完(含 return)才提交 except Exception: conn.rollback() # 任何异常整体回滚, 一行不漏 raise # 异常继续上抛, 绝不吞 finally: conn.autocommit(True) with tx(conn) as cur: cur.execute("UPDATE account SET balance = balance - 50 WHERE id = %s", (uid,))
嵌套事务同理升级: 计数 depth, 只在最外层真正 commit/rollback, 内层 with 复用同一事务。
拿不到锁就等、释放时直接 DEL? TTL 过期后你会删掉别人刚拿到的锁。token + Lua 原子比对才是正解:
import redis from uuid import uuid4 UNLOCK = """if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end""" class DistLock: def __init__(self, rds, key, ttl=30): self.rds, self.key, self.ttl = rds, key, ttl # 只存参数, 不占资源 def __enter__(self): self.token = uuid4().hex # 每次加锁唯一 token if not self.rds.set(self.key, self.token, nx=True, ex=self.ttl): raise LockHeld(self.key) # 拿不到快速失败, 不静默等待 return self def __exit__(self, exc_type, exc_val, tb): # Lua 原子"比对+删除": 只删自己的锁, 防误删他人锁 self.rds.eval(UNLOCK, 1, self.key, self.token) return False # 异常照常传播
分片写文件要同时持有上千个句柄, 手写 try/finally 列表必漏。ExitStack 把"数量未知"变成一行收口:
from contextlib import ExitStack, contextmanager @contextmanager def shard_writers(n=1024): with ExitStack() as stack: # 异常也会全量关闭 files = [stack.enter_context(open(f"out/part-{i:04d}.txt", "w")) for i in range(n)] # 逐个压栈: 1024 个句柄 yield files # 中途崩 → 已开的全部逆序关闭 with shard_writers() as outs: for line in source: # 同 key 恒定落同一分片 outs[hash(line) % len(outs)].write(line) # 离开 with: 哪怕第 700 个分片写入时抛错, 1024 个文件也全部 close
手改 os.environ 再"记得"改回来, 忘一次就污染同进程后续用例, 出偶发红。mock.patch 全家桶本身就是上下文管理器:
from unittest.mock import patch def test_staging_rate_limit(): with patch.dict(os.environ, {"APP_ENV": "staging", "RATE_LIMIT": "10"}), \ freeze_time("2026-09-26"): # 多个 CM 逗号串联 resp = client.get("/api/v1/quotes") assert resp.headers["X-RateLimit-Limit"] == "10" # 断言失败(异常)也会还原环境 — 这就是 with 相比手写 try/finally 的价值
下线清理逻辑到处是 try/except KeyError: pass, 读代码的人却分不清"预期忽略"和"吞异常擦屁股":
from contextlib import suppress def deregister(client_id): with suppress(KeyError): # 只吞 KeyError, 别的类型照抛 del REGISTRY[client_id] # 不存在就当已完成 with suppress(FileNotFoundError): os.unlink(f"/tmp/sock-{client_id}") # 残留 socket 顺手清 metrics.dec("live_clients") # 一定执行 — 语义一眼可读
事故背景: 集成老定价库后 kubectl logs 混进大量裸 print, 日志采集按级别过滤直接失效。不改老库源码的收编法:
import io, logging, contextlib log = logging.getLogger(__name__) def quote_safe(cart): buf = io.StringIO() with contextlib.redirect_stdout(buf): # 进出自动换回真 stdout total = legacy_pricer.quote(cart) # 满屏 print 全部落进 buf if buf.getvalue(): log.info("legacy pricer stdout: %s", buf.getvalue()[:2000]) return total # 老输出变成带级别的结构化日志, 可检索可告警
每个接口手写 t0 = time.time() + finally 打点, 重复且容易漏异常路径。打点本身就该是个 with:
import time, statsd from contextlib import contextmanager @contextmanager def timer(metric, **tags): t0 = time.perf_counter() status = "ok" try: yield except Exception: status = "error"; raise # 失败也打点再抛 finally: ms = (time.perf_counter() - t0) * 1000 statsd.histogram(metric, ms, status=status, **tags) # P95 监控 with timer("pricing.quote", channel="app"): price = quote(cart)
埋点覆盖正常/异常两条路径, 错误率与延迟同一套 tag 维度, Grafana 面板直接分位对比。
裸用 pool.get()/pool.put(), 任何一条提前 return 的分支都是一个泄漏连接。借还对称性交给协议:
class PooledConn: def __enter__(self): self.conn = POOL.get(timeout=5) # 借出发生在 __enter__ return self.conn def __exit__(self, exc_type, exc_val, tb): if exc_type is not None and issubclass(exc_type, OperationalError): POOL.discard(self.conn) # 断连连接销毁, 不还池 else: POOL.put(self.conn) # 归还 — 借必还 return False with PooledConn() as conn: rows = conn.query(SQL, params) # 提前 return 也不会漏还
每个请求 new 一个 client = 每次都做 TCP+TLS 握手。把连接池生命周期钉在 with 上:
import httpx, asyncio limits = httpx.Limits(max_connections=100, max_keepalive_connections=20) async def crawl(urls): async with httpx.AsyncClient(limits=limits, timeout=httpx.Timeout(5.0)) as client: # 为什么挂在 with 上: 池的生命周期 = with 块, 退出时优雅关池 return await asyncio.gather(*(fetch(client, u) for u in urls)) # 一次 TCP+TLS 握手被上千请求摊薄: P50 从 180ms 降到 70ms
每次推理拉新模型包, 解压目录忘了清, 磁盘一周被临时文件吃满。临时目录的正确姿势是一行 with:
import tempfile, tarfile, joblib with tempfile.TemporaryDirectory(prefix="ml-artifact-") as tmp: with tarfile.open(stream=artifact_stream()) as tar: tar.extractall(tmp, filter="data") # 模型包解到临时目录 model = joblib.load(f"{tmp}/model.pkl") return predict(model, features) # 出了 with 目录整棵删除 # 每次拉新工件, 磁盘零残留; 3.12+ 可加 ignore_cleanup_errors=True 兜底
return False 或干脆不写 return。 def __exit__(self, t, v, tb): return True # 错: body 的异常被静默吞掉, 上层"成功" def __exit__(self, t, v, tb): self.close() return False # 对: 只清理, 异常照常上抛
try/except Exception 记日志后吞掉清理错误, return False 保住原异常。 def __exit__(self, t, v, tb): self.conn.close() # 错: close 再抛错 → 顶掉原始异常 def __exit__(self, t, v, tb): try: self.conn.close() except Exception: log.exception("close failed") # 对: 记日志吞掉清理错误 return False # 保住原异常
__enter__, 每次进块重新拿。 class Res: def __init__(self): self.conn = get_conn() # 错: 一创建就占用, 没 with 也占 class Res: def __init__(self): ... def __enter__(self): self.conn = get_conn() # 对: 进块才拿, 出块即还 return self.conn
stack.enter_context(...) 注册 — 任何时刻注册的都保证退出时执行。 @contextmanager def res(): f = open(p) # 错: 这行抛错时 with 未建立 try: yield f # finally 里的清理不会执行 finally: f.close() # 对: with ExitStack() as st: yield st.enter_context(open(p)) # 注册即保证退出 — 任何时刻注册都收尾
def read_cfg(): with open("app.yaml") as f: return f.read() # 对: return 也触发 __exit__, f 必关 # 错: return 前手动 f.close() — 异常路径反而漏关
with a as x, \ b as y: 反斜杠续行, 或改嵌套/ExitStack。 with (open("a") as f, open("b") as g): # 错: 3.9 → SyntaxError process(f, g) with open("a") as f, open("b") as g: # 对: 全版本可用 process(f, g)
with ExitStack() as st: conn = st.enter_context(get_conn()) # 先注册 → 后关 cur = st.enter_context(conn.cursor()) # 后注册 → 先关 # 关键: LIFO — 游标先于连接关闭(依赖正确) # 错: 顺序写反 → 关连接时游标还开着
suppress(KeyError); 用 Exception 也必须圈定在明确"预期失败"的两行内。 with suppress(Exception): # 错: 编程错误/Ctrl+C 也被吞 misspelled_fn() # NameError 无声消失 with suppress(KeyError): # 对: 只吞精确类型 del REGISTRY[cid]
contextlib.asynccontextmanager 或实现 __aenter__/__aexit__。 async with open_lock(): # 错: TypeError: does not support the ... # async context manager protocol async with asyncio.Lock(): # 对: 有 __aenter__/__aexit__ ...
cm = Conn() with cm: work_a() # 错: 复用同一实例, 第二次进块 with cm: work_b() # 覆盖 self.conn, A 的退出关掉 B 的 with Conn(): work_a() # 对: 每次 with 新建实例 with Conn(): work_b()
ExitStack.enter_context(obj) 托管 — 无论后续发生什么, 退出时统一触发。 cm = Conn(); cm.__enter__() # 错: 后面抛错没人调 __exit__ do_work() # → 泄漏且无任何报错 with Conn() as c: # 对: 或 ExitStack.enter_context(cm) do_work()
with lock: # 错: 拿锁做网络 IO, 秒级持锁 data = call_api() # 吞吐退化成串行, P99 飙高 save(data) data = call_api() # 对: 耗时 IO 移出锁外 with lock: save(data) # 锁只罩住读-改-写临界区
ignore_cleanup_errors=True, 或 finally 里 best-effort shutil.rmtree。 with TemporaryDirectory() as d: # 错: 句柄没关就出块 write_files(d) # Windows cleanup → PermissionError with TemporaryDirectory(ignore_cleanup_errors=True) as d: # 对: 3.12+ write_files(d)
def __exit__(self, t, v, tb): self.close() # 不写 return → None → 异常照常传播 # 错: 误以为异常丢了再补 raise → 双重抛出 # 对: 清理型 __exit__ 什么都不返回就对了
yield a, b 绑给 as 的是元组, 当单值用处处 AttributeError。正解: with cm() as (a, b) 解包, 或 yield 一个具名对象/NamedTuple, 调用方更不容易拿错。 @contextmanager def pair(): yield db, cache # as 拿到的是元组 with pair() as p: p.query(...) # 错: AttributeError: 'tuple' object with pair() as (db, cache): # 对: 解包绑定 db.query(...)
__exit__ 只做资源清理这一件事。 def __exit__(self, t, v, tb): if t: refund(self.uid) # 错: 业务补偿埋进协议层, 告警失真 self.close() def __exit__(self, t, v, tb): self.close() # 对: 只清理; 补偿写 body 的 except 里
contextvars(协程/线程本地)代替进程级全局切换。 with redirect_stdout(buf): # 错: body 里库又切了一次, lib.run() # 退出"恢复"的是内层值 → 全局带歪 # 对: 重入场景用 contextvars 按协程/线程隔离 token = cv.set("out") # 各上下文互不干扰
buf = io.StringIO() with redirect_stdout(buf): # 错: sys.stdout 是进程级, run_thread_a() # 线程 B 的 print 也进 buf, 随机丢 # 对: 服务进程统一走 logging, handler 定去向 log.info("done") # 线程安全, 各写各的 appender
with db_tx(conn): — 每次调用产生新的生成器 CM, 这一次调用少不得。 with db_tx: # 错: AttributeError: 'generator' object ... # has no attribute '__enter__' with db_tx(conn): # 对: 调用产生新的生成器 CM ...
async def, 内部清理全部 await; 返回值语义与 __exit__ 相同(True 才吞异常)。 async def __aexit__(self, t, v, tb): self.conn.close() # 错: 返回协程没人等 ... # → RuntimeWarning: coroutine ... never awaited async def __aexit__(self, t, v, tb): await self.conn.close() # 对: 清理动作全部 await