装饰器与上下文管理器

接口超时了，第一反应是给处理函数打一圈耗时：

import time
from collections.abc import Callable
from typing import TypeVar

R = TypeVar("R")


def handle_login(user: str) -> str:
    time.sleep(0.01)
    return f"ok:{user}"


def handle_login_timed(user: str) -> str:
    t0 = time.perf_counter()
    try:
        return handle_login(user)
    finally:
        print(f"handle_login {time.perf_counter() - t0:.3f}s")

每个接口复制这段，改名漏改、try/finally 漏写、有人走了 handle_login 没走包装。高阶函数把「怎么包」收成一次：

def timed(fn: Callable[..., R]) -> Callable[..., R]:
    def inner(*args: object, **kwargs: object) -> R:
        t0 = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            print(f"{fn.__name__} {time.perf_counter() - t0:.3f}s")
    return inner


handle_login = timed(handle_login)
print(handle_login("alice"))

@timed 就是这一行赋值的语法糖。路由表如果按 __name__ 注册，下一行会踩坑：

ROUTES: dict[str, Callable[..., object]] = {}


def route(path: str):
    def deco(fn: Callable[..., R]) -> Callable[..., R]:
        ROUTES[fn.__name__] = fn          # 用函数名当 key，Flask 早期常见
        return fn
    return deco


@route("/login")
@timed
def login(user: str) -> str:
    return f"ok:{user}"


print(login.__name__, list(ROUTES))       # inner  ['inner']

timed 返回的 inner 盖掉了 login。route 登记到的是 "inner"，第二个被 @timed 包的接口再注册，直接把前一个覆盖。functools.wraps 不是礼貌，是身份。基础篇给过最小模型；这里把身份、签名、叠放、方法上的 self、以及 with 的资源协议一次写透。

C++ 没有装饰器，最近的是包装对象 / CRTP / 宏。Go 有中间件 func(http.Handler) http.Handler，和这里的高阶函数是同一件事，只是没有 @ 糖，也不会丢 __name__。Python 把函数当对象改写，糖和坑一起出现。

---

一、装饰器就是高阶函数

1、@ 反糖成一次调用

@deco 完全等价于 f = deco(f)：先把 def f 跑完得到函数对象，立刻交给 deco，绑定名 f 指向返回值。deco 可以返回另一个函数，也可以返回原函数（只登记不包装），也可以返回不可调用的东西——@ 不检查，调用时才爆。

def register(fn):
    fn.registered = True
    return fn                   # 原对象，零包装开销


@register
def ping() -> str:
    return "pong"


print(ping(), ping.registered)

装饰器在 def 执行时跑，不是第一次调用时。模块 import 就会把装饰器跑完。副作用（注册路由、建表、读配置）发生在导入期。循环导入加上装饰器登记，是服务起不来的常见原因：A 装饰器要 B 的表，B 模块还在装。

2、带参装饰器为什么必须三层

@timed 的 timed 接收函数。@route("/login") 先调用 route("/login") 得到一个「接收函数的对象」，再拿这个对象去装饰。两层语义叠在一次 @ 上，所以工厂要再包一层。

from functools import wraps


def repeat(n: int) -> Callable[[Callable[..., R]], Callable[..., R]]:
    if n < 1:
        raise ValueError("n must be >= 1")

    def deco(fn: Callable[..., R]) -> Callable[..., R]:
        @wraps(fn)
        def inner(*args: object, **kwargs: object) -> R:
            last: R | None = None
            for _ in range(n):
                last = fn(*args, **kwargs)
            assert last is not None
            return last
        return inner
    return deco


@repeat(3)
def tick() -> str:
    print("tick")
    return "ok"


print(tick())                   # 打三次 tick，返回 ok

展开：

@repeat(3)          →  tmp = repeat(3)     # 得到 deco
def tick(): ...     →  tick = tmp(tick)    # deco 吃函数

少一层就会把 n 和 fn 抢同一个参数位。有人用默认参数捏成「可带参可不带」：

def timed(fn: Callable[..., R] | None = None, *, unit: str = "s"):
    def deco(f: Callable[..., R]) -> Callable[..., R]:
        @wraps(f)
        def inner(*args: object, **kwargs: object) -> R:
            t0 = time.perf_counter()
            try:
                return f(*args, **kwargs)
            finally:
                dt = time.perf_counter() - t0
                if unit == "ms":
                    dt *= 1000
                print(f"{f.__name__} {dt:.3f}{unit}")
        return inner

    if fn is not None:
        return deco(fn)         # @timed
    return deco                 # @timed(unit="ms")

@timed 走 fn is not None；@timed(unit="ms") 走工厂。括号一旦加上，fn 必须靠关键字，所以 unit 是 keyword-only。这比维护两个名字要省，阅读成本换到「看到括号要想一层」。三层嵌套不是风格问题，是 @expr 只调用一次的直接推论。

C++ 里 template <int N> struct Repeat 用模板参数当 n，实例化期绑定。Go 的中间件工厂 func Repeat(n int) func(http.Handler) http.Handler 和这里同构，只是没有糖，看不漏层。

3、装饰器不是闭包的新语法，闭包是它的实现

inner 能看见 fn 和 n，靠的是 cell，基础篇讲过。这里只补工程后果：装饰器里的可变默认值、共享 cache，和 def f(xs=[]) 是同一个坑。

def memo_wrong(fn, cache={}):   # 所有被装饰函数共享
    @wraps(fn)
    def inner(x):
        if x not in cache:
            cache[x] = fn(x)
        return cache[x]
    return inner

正确：cache 造在 deco(fn) 里，一次装饰一份；或挂在 inner 属性上，便于测试清空。

---

二、functools.wraps：把身份拷回来

1、丢的不只是 __name__

def naked(fn):
    def inner(*args, **kwargs):
        return fn(*args, **kwargs)
    return inner


def documented(x: int) -> int:
    """Add one."""
    return x + 1


wrapped = naked(documented)
print(wrapped.__name__)         # inner
print(wrapped.__doc__)          # None
print(wrapped.__module__)
print(wrapped.__qualname__)     # naked.<locals>.inner
print(wrapped.__annotations__)  # {}

Sphinx 抽 docstring、inspect.signature 做 CLI、FastAPI / Flask 用名字注册、Sentry 用 __qualname__ 聚合错误——全碎。wraps 是 update_wrapper 的偏函数，默认拷这些：

from functools import WRAPPER_ASSIGNMENTS, WRAPPER_UPDATES, wraps, update_wrapper

print(WRAPPER_ASSIGNMENTS)
# 3.13: __module__/__name__/__qualname__/__doc__/__annotate__/__type_params__
# 3.11/3.12 仍含 __annotations__；WRAPPER_UPDATES 是 ('__dict__',)

@wraps(fn)  # 挂在 inner 上：拷 __name__/__doc__/__qualname__/__module__，并设 __wrapped__
def inner(*args: object, **kwargs: object) -> R:
    return fn(*args, **kwargs)

@timed 之后 documented.__name__ 是 "documented"，inspect.signature 是 (x: int) -> int。

__wrapped__ 指向被包函数。inspect.unwrap 沿这条链走到头，inspect.signature 默认也跟。多层装饰器形成 f.__wrapped__.__wrapped__。改签名时至少手动设 __name__ 和 __wrapped__，否则栈里全是 inner。

2、signature 跟 __wrapped__ 走，也跟 __signature__ 走

from inspect import Signature, signature, Parameter


def plus_one(fn):
    @wraps(fn)
    def inner(x: int) -> int:
        return fn(x) + 1
    inner.__signature__ = Signature(
        [Parameter("x", Parameter.POSITIONAL_OR_KEYWORD, annotation=int)]
    )
    return inner


@plus_one
def raw(x: int, y: int) -> int:
    return x + y


print(signature(raw))           # (x: int)，被我们改过

框架（FastAPI 依赖注入、typer CLI）读的是 signature。包装之后参数对不上，运行时 TypeError，或者框架按错名字从 JSON 取值。改了调用约定就要显式改 __signature__，不要指望 wraps 猜你的新签名。

开篇那个路由失败，补上 wraps 之后：@route 在上、@timed 在下（靠近 def），先 timed(login) 得到 __name__ == "login" 的 inner，再 route 按名字登记。顺序反了也行，前提是两边都 wraps。一边漏，另一边按名字登记，就是 ['inner']。

---

三、类装饰器、叠放顺序、装饰方法

1、类当装饰器：靠 __call__ 持有状态

函数装饰器的状态放闭包。状态要暴露、要复用、要继承时，用类：

from collections.abc import Callable
from functools import update_wrapper
from typing import Generic, TypeVar

R = TypeVar("R")


class Counted(Generic[R]):
    def __init__(self, fn: Callable[..., R]) -> None:
        self.fn = fn
        self.calls = 0
        update_wrapper(self, fn)    # 实例也能被 wraps 逻辑拷身份

    def __call__(self, *args: object, **kwargs: object) -> R:
        self.calls += 1
        return self.fn(*args, **kwargs)


@Counted
def add(a: int, b: int) -> int:
    return a + b


print(add(1, 2), add.calls, add.__name__)

@Counted 把 add 绑成 Counted 实例。实例可调用，所以后面对 add(...) 仍成立。update_wrapper(self, fn) 把 __name__ 等拷到实例上。

另一件事是「装饰 class 的函数」：接收 class 对象、返回 class（或替代物）。@dataclass、@total_ordering 走这条路。不要和「类当装饰器」混名。

2、叠放：靠近 def 的先执行

@a
@b
def h(): ...
# 等价
h = a(b(h))

def 时先跑 b 再跑 a；调用时先 a 再 b 再 body。b 离 def 近，先包；a 在最外，先接到调用。洋葱模型：登记类（route）通常在最外，要看见已经 wraps 好的身份；缓存类在最内，贴近原函数。没有统一正确答案，但「谁先谁后」必须是刻意的。日志里只看到 inner，多半是某层没 wraps。

3、装饰方法时 self 怎么传

函数装饰器返回的仍是 function，function 是非数据描述符。s.ping(1)：类字典里是 inner → __get__ 把 s 绑成第一个参数 → inner(s, 1) → fn(s, 1)。self 不需要特殊处理，前提是 inner 用 *args 原样转发。写成 def inner(x) 就会把 self 吃进 x。

装饰器返回的如果是已经绑定好的对象（比如某个实例的 bound method），描述符协议被绕开，self 丢。类装饰器里 __get__ 要自己转发给原函数：

from types import MethodType


class Timed:
    def __init__(self, fn: Callable[..., R]) -> None:
        self.fn = fn
        update_wrapper(self, fn)

    def __call__(self, *args: object, **kwargs: object) -> R:
        return self.fn(*args, **kwargs)

    def __get__(self, obj: object, owner: type | None = None):
        if obj is None:
            return self
        return MethodType(self, obj)     # 绑定后调用 Timed.__call__(self_timed, obj, ...)


class Svc2:
    @Timed
    def ping(self, x: int) -> int:
        return x + 1


print(Svc2().ping(1))

漏写 __get__ 时，Svc2().ping(1) 等于 Timed.__call__(timed_instance, 1)，原函数的 self 丢了，典型报错是 ping() missing 1 required positional argument。函数装饰器没有这个问题，因为返回值仍是 function，function 是非数据描述符。类装饰器要当方法用，自己实现 __get__。

4、classmethod / staticmethod 的顺序

两者本身就是描述符。装饰器叠在它们外面还是里面，决定你拿到的是裸函数还是描述符对象。

def trace(fn):
    @wraps(fn)
    def inner(*args, **kwargs):
        print("trace", fn.__name__, args)
        return fn(*args, **kwargs)
    return inner


class M:
    @classmethod
    @trace
    def via_cls(cls, x: int) -> tuple[type, int]:
        return cls, x

    @trace
    @classmethod
    def via_trace_outer(cls, x: int) -> tuple[type, int]:
        return cls, x


print(M.via_cls(1))             # trace 看到 (M, 1)，正确
try:
    print(M.via_trace_outer(1))
except TypeError as e:
    print("outer trace:", e)

@classmethod 在上、@trace 在下（靠近 def）：classmethod 接到的是 inner，调用时先绑定 cls 再进 inner，cls 在 args[0]。反过来：trace 接到一个 classmethod 对象，inner 调用它时不会走描述符绑定，cls 没被传进去。

规则：classmethod / staticmethod / property 放最外（离 def 最远），自定义包装放里面。3.9 起 classmethod 自己会转发 __wrapped__ 等属性，但顺序仍不能反。property 更严：@property 下面才是 fget，再下面才是你的 deco。

class P:
    @property
    @trace
    def n(self) -> int:
        return 42


print(P().n)                    # 访问属性时进 trace

写成 @trace @property 会试图调用 trace(<property object>)，inner 不是 getter。staticmethod 同一条规则：描述符必须在最外，否则绑定不发生。

---

四、缓存、重试、权限、超时

四个都是「在调用前后插一刀」。差别在副作用和失败语义。

1、缓存：键必须盖住语义，失效必须可见

from functools import wraps, lru_cache
import threading


def memoize(fn: Callable[..., R]) -> Callable[..., R]:
    cache: dict[tuple[object, ...], R] = {}
    lock = threading.Lock()

    @wraps(fn)
    def inner(*args: object, **kwargs: object) -> R:
        key = (args, tuple(sorted(kwargs.items())))
        with lock:
            if key in cache:
                return cache[key]
        result = fn(*args, **kwargs)
        with lock:
            cache[key] = result
        return result

    inner.cache = cache  # type: ignore[attr-defined]
    return inner


@lru_cache(maxsize=128)
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)


print(fib(20), fib.cache_info())

生产优先 functools.lru_cache / cache（3.9+）。自己写只为了看清：未命中时的计算在锁外，避免把慢函数串行化；代价是可能 stampede。key 必须可哈希，否则 TypeError。

缓存装饰器默认假设纯函数。fn 读当前时间、读 DB、读配置，缓存会把第一次的世界冻住。对象方法上 @lru_cache 会把 self 放进 key，实例永远退不了缓存——轻则泄漏，重则逻辑绑在已死对象上。方法缓存用 cached_property，或 key 里不放 self 只放 id / 业务主键。

2、重试：只对幂等操作打开

class TransientError(Exception):
    pass


def retry(
    *,
    times: int = 3,
    delays: Iterable[float] = (0.05, 0.1, 0.2),
    on: type[BaseException] | tuple[type[BaseException], ...] = TransientError,
) -> Callable[[Callable[..., R]], Callable[..., R]]:
    delay_list = list(delays)

    def deco(fn: Callable[..., R]) -> Callable[..., R]:
        @wraps(fn)
        def inner(*args: object, **kwargs: object) -> R:
            last: BaseException | None = None
            for attempt in range(times):
                try:
                    return fn(*args, **kwargs)
                except on as e:
                    last = e
                    if attempt == times - 1:
                        break
                    wait = delay_list[min(attempt, len(delay_list) - 1)]
                    time.sleep(wait)
            assert last is not None
            raise last
        return inner
    return deco


@retry(times=5, on=TransientError)
def fetch_id(n: int) -> int:
    # 幂等：相同 n 永远代表「读同一条」。重试安全。
    if random.random() < 0.6:
        raise TransientError("flaky")
    return n


print(fetch_id(7))

幂等：同一请求做一次和做多次，对系统的可观察结果相同。GET 资源、PUT 全量覆盖、带幂等键的支付查询，可以重试。POST /orders 每次造新单、DELETE 已经有副作用的转账，不能无脑重试。装饰器看不见业务，它只会把函数再调一遍。把 @retry 贴到非幂等接口上，故障时加倍下单。

还要限制 on：只捕瞬时错误（超时、连接重置、429）。捕 Exception 会把 ValueError / 权限错误也重试，把 bug 放大成延迟。times 是总尝试次数不是失败次数。延迟用已有序列，不要在装饰器默认参数里放 [] 再 append。

Go 里这段通常是客户端中间件。Python 的危险在于贴 @retry 太便宜，贴错的成本在账上。

3、权限：失败要在调用前，不要在副作用后

class PermissionDenied(Exception):
    pass


def require(*perms: str):
    def deco(fn: Callable[..., R]) -> Callable[..., R]:
        @wraps(fn)
        def inner(*args: object, **kwargs: object) -> R:
            ctx = kwargs.get("ctx") or (args[0] if args else None)
            granted = set(getattr(ctx, "perms", ()))
            missing = [p for p in perms if p not in granted]
            if missing:
                raise PermissionDenied(f"missing {missing}")
            return fn(*args, **kwargs)
        return inner
    return deco

鉴权必须在任何写操作之前。装饰器保证这一点，比在函数第一行手写 if 难漏。ctx 怎么传到 inner 要有约定：明确参数名，不要靠 args[0] 猜。

4、超时：线程里的超时是假的，进程/异步才是真的

import signal


class Timeout(Exception):
    pass


def time_limit(seconds: float):
    def deco(fn: Callable[..., R]) -> Callable[..., R]:
        @wraps(fn)
        def inner(*args: object, **kwargs: object) -> R:
            def _handler(signum: int, frame: object) -> None:
                raise Timeout(f"{fn.__name__} exceeded {seconds}s")

            if not hasattr(signal, "SIGALRM"):
                raise NotImplementedError("SIGALRM not available")
            old = signal.getsignal(signal.SIGALRM)
            signal.setitimer(signal.ITIMER_REAL, seconds)
            signal.signal(signal.SIGALRM, _handler)
            try:
                return fn(*args, **kwargs)
            finally:
                signal.setitimer(signal.ITIMER_REAL, 0)
                signal.signal(signal.SIGALRM, old)
        return inner
    return deco

SIGALRM 只在主线程、Unix 上有意义；Windows 没有。它也不能打断正在 C 里阻塞的调用。线程里用它会把信号送到主线程，杀错人。生产超时：

• 同步 HTTP：requests / httpx 的 timeout=，发生在客户端，不靠装饰器。
• 线程池任务：Future.result(timeout=)，到期你拿到异常，工作线程还在跑，只是你不等了。
• 真取消：asyncio.wait_for / asyncio.timeout（3.11+），协作式取消。

装饰器 time_limit 适合约束纯计算、演示协议，不适合当 RPC 超时的实现。写进服务里要标明平台限制，否则 Linux 能跑、容器里多线程一上就 mystery。

---

五、上下文管理器：with 的反糖

1、__enter__ / __exit__

with cm as x:
    body

展开：

mgr = cm
x = mgr.__enter__()
exc = True
try:
    body
except:
    exc = False
    if not mgr.__exit__(*sys.exc_info()):
        raise
finally:
    if exc:
        mgr.__exit__(None, None, None)

__enter__ 的返回值绑到 as 后面。__exit__(exc_type, exc, tb) 在离开 with 无论是否异常都会被调用。返回真值（True）表示「异常已处理」，解释器不再往上抛；返回 False / None 继续传播。

C++ 的 RAII：析构函数在对象离开作用域时跑，异常路径由栈展开保证。Go 的 defer：函数返回时按 LIFO 跑，不是词法块结束。Python 的 with 对齐的是 C++ 作用域，不是 Go defer。先把 __exit__ 的返回值钉死。

2、返回 True 会吞异常

class Swallow:
    def __enter__(self) -> None:
        return None

    def __exit__(self, exc_type, exc, tb) -> bool:
        if exc_type is ValueError:
            print("swallowed", exc)
            return True
        return False


with Swallow():
    raise ValueError("bad")
print("still running")

try:
    with Swallow():
        raise TypeError("nope")
except TypeError as e:
    print("propagated", e)

contextlib.suppress 就是一个只吞指定类型、__exit__ 返回 True 的管理器。自己写 __exit__ 时默认 return False 或不写 return。误 return True 会把业务异常吃掉，调用方以为成功。审查 __exit__ 和审查 except Exception: pass 同等力度。

__exit__ 自己再抛，原异常被挂到 __context__ 上（或 3.x 的隐式链）。要替换异常用 raise New from exc。不要在 __exit__ 里静默失败。

3、__enter__ 失败不会调 __exit__

资源是 __enter__ 拿到的。__enter__ 里 raise OSError，__exit__ 不会跑，没有东西可清。构造函数里 open、__enter__ 只 return self 的写法，会让构造失败和 with 失败走两条清理路径——把申请放进 __enter__。

---

六、contextlib：生成器写出 CM，以及一组工具

1、@contextmanager：yield 这边是进入，那边是退出

from contextlib import contextmanager
from collections.abc import Iterator


@contextmanager
def timed_block(name: str) -> Iterator[None]:
    t0 = time.perf_counter()
    try:
        yield
    finally:
        print(f"{name} {time.perf_counter() - t0:.4f}s")

yield 前 = __enter__，yield 的值 = as 绑定，yield 后 = __exit__。try/finally 必须写，否则 with 体内异常会跳过清理。@contextmanager 会把体内异常在 yield 点重新抛进生成器；你不捕，它继续向外走，同时 finally 仍跑。捕了还想让外层看见，再 raise。上一篇的生成器 yield 产出元素；这里的 yield 产出一次控制权。同一个关键字，两个协议。yield 前申请、忘记 finally，异常路径泄漏——和生成器持锁是同一类 bug。

手写 __enter__/__exit__ 适合对象本身就是资源（文件、连接、锁）。@contextmanager 适合「一段代码前后各一刀」、不想为它单开一个类。

2、closing / suppress / redirect_stdout

from contextlib import closing, suppress, redirect_stdout
from io import StringIO

with suppress(FileNotFoundError):
    open("/no/such/file")

buf = StringIO()
with redirect_stdout(buf):
    print("hello")
print(buf.getvalue())

closing(obj) 在离开 with 时调 obj.close()，给「有 close 但不是 CM」的对象用。已经实现了 __enter__ 的不要再套。

suppress 是显式吞异常。只用于「缺省文件、可选锁、探测性 import」这类确定可忽略的路径，不要当通用 try/except。

redirect_stdout / redirect_stderr 改的是 sys.stdout，线程间互相踩。测一个打印函数可以；服务进程里不要拿它做日志采集。

3、ExitStack：动态数量的 with

with a, b, c: 数量要在写代码时定死。运行期才知道开几个文件、租几条连接，用 ExitStack：

from contextlib import ExitStack
from pathlib import Path


def concat_files(paths: list[Path], dest: Path) -> None:
    with ExitStack() as stack:
        files = [stack.enter_context(p.open(encoding="utf-8")) for p in paths]
        out = stack.enter_context(dest.open("w", encoding="utf-8"))
        for f in files:
            out.write(f.read())
        # 离开 with：按 LIFO close。某个 close 抛，其余仍会 close

callback 注册任意清理，LIFO：后进入的先退出。和 C++ 局部变量析构、Go defer 一致。pop_all() 把栈挪走，用于「构造失败则全退、构造成功则所有权交给对象」：构造里 try: stack.enter_context(...) ，except: stack.close(); raise，成功则把 stack 存成成员，对象自己当 CM 时 __exit__ 再 close。这就是连接池租约的骨架。

---

七、可重入 vs 不可重入：锁、文件、租约

1、锁

import threading

lock = threading.Lock()         # 不可重入
rlock = threading.RLock()       # 可重入，同一线程可多次 acquire


def outer() -> None:
    with lock:
        inner()                 # 死锁：Lock 不可重入


def inner() -> None:
    with lock:
        pass

Lock 同一线程第二次 acquire 会永远等自己。RLock 记 owner 和计数，配对 release 到零才放开。@synchronized 若内部 with self._lock、方法互相调用，必须用 RLock，否则自死锁。代价：RLock 更重，也拆掉了「临界区内不会再进同一把锁」的证明。

with lock: 在 __enter__ 里 acquire，__exit__ 里 release，异常也会放。手写 acquire/release 漏在 return 前面就是泄漏。Go 没有 RAII，defer mu.Unlock() 绑在函数返回；临界区比函数短时要再包一层函数。Python 用 with 把临界区收成块。

2、文件

文件对象的 close 可多次调用，第二次是空操作。with open(...) as f 离开就关，不管 return / 异常。嵌套 with 打开同一路径是两个 fd，不是可重入。可重入是「同一个 CM 实例进两次」——自己用深度计数实现，默认 open 没有这个计数。不要假设方法 A 里 with self.file、A 调 B、B 再 with self.file 能活。with 应该包住所有权边界。

3、连接池租约

import queue


class Conn:
    def __init__(self) -> None:
        self.closed = False

    def close(self) -> None:
        self.closed = True


class Lease:
    def __init__(self, conn: Conn, pool: "SimplePool") -> None:
        self.conn = conn
        self._pool = pool
        self._released = False

    def __enter__(self) -> Conn:
        return self.conn

    def __exit__(self, exc_type, exc, tb) -> None:
        if isinstance(exc, OSError):
            self._released = True          # 坏连接丢弃，不还池
            return
        if not self._released:
            self._pool._q.put(self.conn)
            self._released = True


class SimplePool:
    def __init__(self, n: int) -> None:
        self._q: queue.Queue[Conn] = queue.Queue()
        for _ in range(n):
            self._q.put(Conn())

    def acquire(self) -> Lease:
        return Lease(self._q.get(), self)


with SimplePool(2).acquire() as conn:
    print(conn.closed)

租约的 __exit__ 决定还是丢。异常类型要挑：业务 ValueError 连接还是好的，该还；OSError / 协议损坏该丢。漏 __exit__（生成器里 yield conn 没 finally）会把池子抽干，下一次 acquire 阻塞到超时。上一篇的「生成器持有锁」，这里是「生成器持有租约」，同一类 bug。

不可重入：同一条连接不能同时被两个 with 借出。池子用队列保证；如果连接自己不是线程安全的，租约还要保证不同时进入。可重入连接（引用计数、同一线程多次 acquire 同一条）是另一套实现，默认不要做，除非 API 明确。

---

八、和 C++ RAII、Go defer 对照

三套释放模型：

	C++ RAII	Go defer	Python with
触发点	对象离开词法作用域（栈展开）	函数返回（含 panic）	离开 with 块
顺序	后构造先析构	LIFO	__exit__ LIFO（多 CM / ExitStack）
异常	析构不该再抛	defer 里 recover	__exit__ 返回 True 吞掉
动态数量	vector<unique_ptr>	for 里 defer（小心循环变量）	ExitStack
函数中段释放	开子作用域 { lock_guard g(m); }	必须再包函数，或手动 Unlock	with 块想多短就多短

Python 没有 defer。在函数开头 f = open(...) 指望函数结束时关，不会关。try/finally 是手动 RAII；with 是它的协议化。把 Go 的「函数入口 defer 一排」搬过来，会把临界区拉到整个函数——锁粒度、文件持有时间都错。

def go_style_wrong(path: str) -> int:
    f = open(path, encoding="utf-8")        # 没有 defer f.close()
    for i, line in enumerate(f, 1):
        if "STOP" in line:
            return i                        # 泄漏
    f.close()
    return i


def cpp_style(path: str) -> int:
    with open(path, encoding="utf-8") as f: # 词法作用域 = 文件寿命
        for i, line in enumerate(f, 1):
            if "STOP" in line:
                return i                    # __exit__ 仍跑
        return i

return 在 with 里，__exit__ 照样跑，和 C++ return 时局部 ifstream 析构一样。Go 也保证 defer 在 return 时跑，但 defer 看不到你在循环中间想关文件——它等函数结束。要中途关，Go 写 func() { f.Close() }() 或把循环拆出去；Python 让 with 块在循环外或循环内，由你选作用域。

装饰器和 CM 经常叠：装饰器改每次调用的前后；CM 改一段词法区域的前后。装饰器内部用 with（@traced 里 with span(fn.__name__)），不要反过来在 __enter__ 里去改函数对象。计时、锁、租约、事务，能写成 CM 的先写成 CM，再按需包一层装饰器。

with 不是垃圾回收。循环引用、忘记离开块、生成器冻住 with 里的 yield，资源都不会在你以为的时间点释放。协议的价值是把释放从「对象没人引用了」提前到「这块代码结束了」。C++ 靠类型系统强制；Go 靠程序员记 defer；Python 靠 with 和审查。