跳转至

上下文管理器:with 到底在做什么

各位先来看一段代码,猜猜哪里有坑:

f = open('/tmp/two_drops_of_water.txt', 'w')
f.write('两点水的日记:今天打卡迟到了三分钟。')
f.close()

「这有什么问题?文件打开了、写完了、关掉了,一气呵成。」善于思考的你可能会这么说。

可是,万一 f.write 那一行抛了异常呢?

close 永远不会被执行,文件描述符就这么悬在那儿。一两次没事,可如果是个 Web 服务跑一整天,几千个请求里只要有几十个写文件失败,操作系统的 fd 就会被慢慢耗光,最后报一个看起来八竿子打不着的「Too many open files」。这种 bug,定位起来非常痛苦。

那怎么办?老办法当然是 try/finally

f = open('/tmp/two_drops_of_water.txt', 'w')
try:
    f.write('两点水的日记:今天打卡迟到了三分钟。')
finally:
    f.close()

这样写虽然稳,但是丑。一个简单的写文件,外面要套一层 try/finally ,缩进就多一级。如果接下来还要打开第二个文件、第三个文件,缩进会一层一层套下去,很快就会写出「金字塔代码」。

童鞋们一定见过 Python 的另一种写法:

with open('/tmp/two_drops_of_water.txt', 'w') as f:
    f.write('两点水的日记:今天打卡迟到了三分钟。')

短短两行,没有 try ,没有 finally ,没有手动 close ,但是不管 write 是否抛异常,文件都会被正确关闭。

这个 with 到底是个什么东西?为什么它能做到这么神奇的事情?这一章我们就把它从里到外讲清楚。

with 的本质:上下文管理协议

with 后面跟的不是「文件」,也不是「连接」、「锁」之类的具体东西,而是一类东西,Python 给它起了个名字,叫「上下文管理器」(context manager)。

什么样的对象能算上下文管理器?只要满足两个方法:

  • __enter__(self) :进入 with 块的时候被调用,返回值会赋给 as 后面的变量
  • __exit__(self, exc_type, exc_val, tb) :离开 with 块的时候被调用,无论是正常离开还是异常离开

这就是「上下文管理协议」(context manager protocol)。它的全部规则就这两条,没有第三条。

我们可以拿打卡这个老朋友来手写一个最简单的上下文管理器,让各位先有个直观感受:

class Punch:
    def __enter__(self):
        print('两点水进入打卡区')
        return self

    def __exit__(self, exc_type, exc_val, tb):
        print('两点水离开打卡区')


with Punch() as p:
    print('两点水正在工位摸鱼')

输出:

两点水进入打卡区
两点水正在工位摸鱼
两点水离开打卡区

各位看到了什么?with Punch() as p 这一行做了三件事:

  1. 创建一个 Punch() 实例
  2. 调用这个实例的 __enter__() 方法
  3. __enter__() 的返回值赋给 p

然后才执行 with 块里面的代码。等 with 块结束(不管是正常结束还是异常结束),就调用 __exit__()

这个流程看着是不是很眼熟?没错,它本质上就是把:

p_obj = Punch()
p = p_obj.__enter__()
try:
    print('两点水正在工位摸鱼')
finally:
    p_obj.__exit__(None, None, None)

这一坨样板代码,藏到了 with 这个语法糖背后。with 一行顶六行,就是这么来的。

顺手回头看一眼 open

弄懂了协议,我们再回头看 open()

with open('/tmp/two_drops_of_water.txt', 'w') as f:
    f.write('一行字。')

open() 返回的文件对象本身就实现了 __enter____exit__ 。它的 __enter__ 直接 return self ,所以 f 拿到的就是文件对象本身;它的 __exit__ 里调用的就是 close() 。简简单单,但是覆盖了所有的异常路径。

各位可以自己验证一下:

f = open('/tmp/two_drops_of_water.txt', 'w')
print(hasattr(f, '__enter__'))
print(hasattr(f, '__exit__'))
f.close()

输出:

True
True

不光是文件,标准库里的 socketthreading.Locksqlite3.Connection ,第三方库里的 requests.Sessionopen(...).fileobj 等等,几乎所有「需要成对操作」的资源都实现了上下文管理器协议。打开/关闭、获取/释放、连接/断开,凡是这种「有借有还」的事情,扔给 with 去管准没错。

我们再多看一个例子,体会一下上下文管理器是怎么把「成对操作」的细节藏起来的。threading.Lock 这个类大家应该不陌生,多线程编程里几乎离不开它:

import threading


lock = threading.Lock()

# 不用 with 的写法,需要手动 acquire 和 release
lock.acquire()
try:
    print('两点水正在临界区里搬砖')
finally:
    lock.release()

# 用 with 的写法,Python 帮我们把 acquire 和 release 都安排好了
with lock:
    print('两点水又一次进入临界区')

输出:

两点水正在临界区里搬砖
两点水又一次进入临界区

注意这里 with lock: 没有写 as ... 。因为锁的 __enter__ 返回的就是它自己,对绝大多数使用场景没用——我们关心的是「锁住了」这个状态,不是返回值本身。所以 as 可以省。这也是大部分锁、信号量、事件之类的上下文管理器的常见写法。

一个稍微有用一点的例子:打卡上下文

光是 print 几行字太单薄了,我们让这个 Punch 做点正经事——记录员工的进入时间和离开时间,并把它写到「日志」里:

import time


class PunchLog:
    def __init__(self, name):
        self.name = name
        self.records = []

    def __enter__(self):
        self.start = time.time()
        self.records.append(f'{self.name} 进入工位')
        return self

    def __exit__(self, exc_type, exc_val, tb):
        cost = time.time() - self.start
        self.records.append(f'{self.name} 离开工位,停留 {cost:.4f} 秒')


log = PunchLog('两点水')
with log:
    time.sleep(0.05)
    print('正在划水……')

for line in log.records:
    print(line)

输出大致是:

正在划水……
两点水 进入工位
两点水 离开工位,停留 0.0507 秒

注意 __enter__ 这次没有 return self 也没关系,因为我们外面用的是 with log: ,不需要 as 接返回值,log 这个变量本身就是上下文管理器。如果改成 with log as l: ,那 l 会变成 None ,因为 __enter__ 默认返回 None 。这是新手最容易踩的小坑:「我的 as x 怎么是 None ?」八成是 __enter__ 忘了 return

同时管理多个上下文

实际项目里,经常需要同时打开两个甚至三个文件——读一个、写一个、再写一个错误日志。早期 Python 只能这么写:

with open('/tmp/in.txt', 'w') as fin:
    pass

with open('/tmp/in.txt') as fin:
    with open('/tmp/out.txt', 'w') as fout:
        fout.write(fin.read())

两层缩进。如果再加一个 err.log ,就是三层。

后来 Python 允许在一个 with 里写多个上下文,用逗号分隔:

with open('/tmp/in.txt') as fin, open('/tmp/out.txt', 'w') as fout:
    fout.write(fin.read())

这样比嵌套写法清爽多了。可是问题又来了——如果上下文很多,一行会变得贼长,没法折行。

Python 3.10 给出了一个新语法,叫 PEP 617 「带括号的上下文管理器」,允许把多个上下文用括号包起来:

with (
    open('/tmp/in.txt') as fin,
    open('/tmp/out.txt', 'w') as fout,
):
    fout.write(fin.read())

注意三件事:

  1. 括号里每一行后面都可以加逗号,包括最后一个,PEP 617 允许尾随逗号
  2. 这个语法是 Python 3.10+ 才有的
  3. 代码风格上,这是处理「同时管理 3 个以上上下文」时最推荐的写法

如果各位还在用 3.9 或者更老的版本,要么用逗号分隔的单行写法,要么后面我们会讲到 ExitStack

顺手提一个细节:多个上下文管理器是按从左到右的顺序依次 __enter__ ,按从右到左的顺序依次 __exit__ 。这是个非常重要的语义保证——比如「先拿锁再开文件」的代码,退出时一定是「先关文件再放锁」。我们写一个小 demo 验证一下:

from contextlib import contextmanager


@contextmanager
def stage(name):
    print(f'{name} __enter__')
    try:
        yield name
    finally:
        print(f'{name} __exit__')


with stage('A') as a, stage('B') as b, stage('C') as c:
    print(f'with 块内部:{a} {b} {c}')

输出:

A __enter__
B __enter__
C __enter__
with 块内部:A B C
C __exit__
B __exit__
A __exit__

各位看到了,C 是最后进入的,也是第一个退出的。和函数调用栈、嵌套 try/finally 的语义完全一致。

exit 的三个参数:异常处理

前面有提到,__exit__ 的签名是 __exit__(self, exc_type, exc_val, tb) ,但一直没解释这三个参数干嘛用。

正常情况下,with 块里没抛异常,这三个参数都是 None 。一旦出了异常,它们分别是:

  • exc_type :异常的类型(比如 ZeroDivisionError
  • exc_val :异常的实例
  • tb :traceback 对象

更关键的是 __exit__ 的返回值:

  • 返回 True (或者任何「真值」):异常被「吞掉」,不再往外抛
  • 返回 FalseNone 、或者根本不写 return :异常继续往外抛

我们写一个能把 ZeroDivisionError 吞掉的上下文管理器看看:

class IgnoreZeroDivision:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, tb):
        if exc_type is ZeroDivisionError:
            print(f'吞掉了一个除零异常:{exc_val}')
            return True
        return False


with IgnoreZeroDivision():
    x = 1 / 0
    print('这句话不会被打印,因为上一行抛异常了')

print('但是这句话会被打印,因为异常被 __exit__ 吞掉了')

输出:

吞掉了一个除零异常:division by zero
但是这句话会被打印,因为异常被 __exit__ 吞掉了

如果 with 块里抛的是 ZeroDivisionError 之外的异常,比如 ValueError__exit__ 会返回 False ,异常照常往外冒。

需要强调一点:吞掉异常这个能力非常强,但请务必谨慎使用。一般来说,上下文管理器的职责是「无论怎样都把资源清理掉」,而不是「假装异常没发生过」。盲目地 return True 可能掩盖真正的 bug 。标准库里真正会吞异常的上下文管理器寥寥无几,最常见的就是后面要讲的 contextlib.suppress

还有一个常见的错觉:「我能不能在 __exit__ 里把异常换成另一种异常抛出来?」答案是可以,但是有讲究。直接在 __exit__raise 一个新异常会把当前正在传播的异常「替换掉」,这种情况下原异常会变成新异常的 __context__ 。我们写个例子让童鞋们看清楚:

class TranslateException:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, tb):
        if exc_type is not None:
            raise RuntimeError(f'我把 {exc_type.__name__} 翻译成了 RuntimeError') from exc_val


try:
    with TranslateException():
        1 / 0
except RuntimeError as e:
    print(f'外面拿到的是:{e}')
    print(f'原因是:{e.__cause__}')

输出:

外面拿到的是:我把 ZeroDivisionError 翻译成了 RuntimeError
原因是:division by zero

这种「翻译异常」的模式在写库的时候特别有用。比如数据库驱动里把底层各种网络错误统一翻译成 DatabaseError ,外层就不用关心是 socket 断了、还是 SSL 握手失败、还是协议解析出错了。raise X from Y 这个语法在异常处理章节里讲过,这里复习一下用法即可。

contextlib:标准库的好朋友

写一个上下文管理器,只为了管几行代码,居然要写一个完整的类、写两个 dunder 方法,是不是有点重?

Python 标准库里有个叫 contextlib 的模块,专门给我们提供了一堆工具来简化这件事。最重要的就是 @contextmanager 这个装饰器。

各位还记得装饰器章节里提到的生成器吗?@contextmanager 就是把一个生成器函数变成上下文管理器。它的规则非常简单:

  • yield 之前的代码 = __enter__
  • yield 出来的值 = __enter__ 的返回值
  • yield 之后的代码 = __exit__

我们把上面那个 PunchLog@contextmanager 重写一遍,对比一下:

import time
from contextlib import contextmanager


@contextmanager
def punch_log(name):
    records = []
    start = time.time()
    records.append(f'{name} 进入工位')
    try:
        yield records
    finally:
        cost = time.time() - start
        records.append(f'{name} 离开工位,停留 {cost:.4f} 秒')


with punch_log('两点水') as records:
    time.sleep(0.05)
    print('继续划水……')

for line in records:
    print(line)

输出:

继续划水……
两点水 进入工位
两点水 离开工位,停留 0.0506 秒

是不是清爽多了?类一下子变成了一个函数,两个 dunder 方法变成了一个 yield

注意里面那个 try/finally 。为什么需要它?

因为 with 块里面可能抛异常,异常会从 yield 那一行重新「钻」回生成器函数里。如果不写 try/finally ,那异常会把后面的清理代码直接绕过去,资源就泄露了。try/finally 才能保证「不管 yield 抛不抛异常,finally 里的清理代码一定会执行」。

这是用 @contextmanager 写上下文管理器时最重要的一条铁律:清理代码要包在 finally 里。新手最常见的 bug 就是不写 try/finally ,结果异常路径下资源没释放。

@contextmanager + 异常处理

那如果我想吞掉异常呢?用 @contextmanager 怎么写?

很简单,把 yield 包在 try/except 里就行:

from contextlib import contextmanager


@contextmanager
def ignore_zero_division():
    try:
        yield
    except ZeroDivisionError as e:
        print(f'用 @contextmanager 吞掉了:{e}')


with ignore_zero_division():
    x = 1 / 0
    print('这一行不会执行')

print('但是这一行会')

输出:

用 @contextmanager 吞掉了:division by zero
但是这一行会

except 捕获到异常之后,函数正常返回,对应的就是 __exit__ 返回 True ;如果异常没被捕获、被生成器原样抛出来,对应的就是 __exit__ 让异常继续向上传播。@contextmanager 内部帮你把这两条路接好了,我们只需要按 Python 异常处理的常规思路来写就行。

contextlib 里的几个常客

contextlib 不只有 @contextmanager ,还有几个开箱即用的小工具,用过几次之后会觉得离不开:

closing:把没实现 with 的对象包成上下文

不是所有需要 close 的对象都实现了上下文管理器协议。最典型的就是 urllib.request.urlopen 在某些老版本里返回的对象,以及一些自定义的资源对象。closing 帮我们补上这一层:

from contextlib import closing


class Connection:
    def __init__(self, name):
        self.name = name

    def close(self):
        print(f'{self.name} 已关闭')


with closing(Connection('数据库连接')) as conn:
    print(f'正在使用 {conn.name}')

输出:

正在使用 数据库连接
数据库连接 已关闭

closing(obj) 等价于一个上下文管理器,它的 __enter__ 返回 obj ,它的 __exit__ 调用 obj.close() 。简单粗暴。

suppress:优雅地忽略异常

from contextlib import suppress


with suppress(FileNotFoundError):
    open('/tmp/this_file_definitely_does_not_exist_12345.txt').read()

print('哪怕文件不存在,我也能继续往下走')

输出:

哪怕文件不存在,我也能继续往下走

suppress 就是上面 IgnoreZeroDivision 的标准库通用版。它接受任意多个异常类型,匹配上就吞掉,不匹配就往上抛。

suppresstry: ... except SomeError: pass 短,而且更明显地表达「我就是要忽略这个异常」的意图。但同样请谨慎使用,不要用它来掩盖真 bug 。

redirect_stdout / redirect_stderr:临时改写标准输出

有时候我们想把某段代码的 print 输出抓到字符串里,或者写到文件里,又不想动它的源码:

import io
from contextlib import redirect_stdout


buf = io.StringIO()
with redirect_stdout(buf):
    print('这段话不会出现在终端')
    print('而是被抓到了 buf 里')

captured = buf.getvalue()
print('我抓到了:')
print(captured)

输出:

我抓到了:
这段话不会出现在终端
而是被抓到了 buf 里

测试代码、或者把第三方库的日志重定向到自己的 logger 时,这两个工具特别好使。

nullcontext:什么都不做的占位上下文

考虑这种场景:你的函数有时候需要在锁里跑,有时候不需要。怎么写得优雅一点?

from contextlib import nullcontext
import threading


def do_stuff(lock=None):
    cm = lock if lock is not None else nullcontext()
    with cm:
        print('做事')


lock = threading.Lock()
do_stuff(lock)
do_stuff()

输出:

做事
做事

nullcontext()__enter____exit__ 啥都不做,纯粹是为了让 with 这一行能写得通用。你不用再去写 if lock: with lock: ... else: ... 这种又臭又长的分支。

nullcontext 还能携带一个值,假装它就是 __enter__ 的返回结果——这在「有时候已经有现成的对象,有时候要新打开一个」的场景里非常顺手:

from contextlib import nullcontext


def read_text(source):
    cm = nullcontext(source) if hasattr(source, 'read') else open(source)
    with cm as f:
        return f.read()


import io
text1 = read_text(io.StringIO('我是一个已经打开的对象'))
print(text1)

with open('/tmp/two_drops_e.txt', 'w') as f:
    f.write('我是写在磁盘上的内容')

text2 = read_text('/tmp/two_drops_e.txt')
print(text2)

输出:

我是一个已经打开的对象
我是写在磁盘上的内容

一个函数,既能接受字符串路径,也能接受已经打开的文件对象,写法一气呵成。这种「适配两种输入」的小套路在工程里非常常用。

实战:临时改环境变量

来看一个非常常见的需求:测试一段代码在某个环境变量被设置时的行为,但是又不想污染当前进程的环境。

各位会怎么写?

import os
from contextlib import contextmanager


@contextmanager
def set_env(**kwargs):
    old = {}
    for k, v in kwargs.items():
        old[k] = os.environ.get(k)
        os.environ[k] = v
    try:
        yield
    finally:
        for k, v in old.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v


os.environ.pop('TWO_DROPS_OF_WATER', None)
print('进入前:', os.environ.get('TWO_DROPS_OF_WATER'))

with set_env(TWO_DROPS_OF_WATER='hello'):
    print('with 里:', os.environ.get('TWO_DROPS_OF_WATER'))

print('退出后:', os.environ.get('TWO_DROPS_OF_WATER'))

输出:

进入前: None
with 里: hello
退出后: None

各位看仔细:进入 with 之前,把要修改的环境变量的「原值」记下来;with 块里,把它们改成新值;退出的时候——不管是不是正常退出——把原值恢复回去。这就是上下文管理器最经典的用法:「临时改一下,离开时还原」。

类似的模式还能用在「临时切换工作目录」、「临时改 sys.path」、「临时调高日志级别」等场景。一旦掌握了这个套路,写起来手感非常顺。

临时切换工作目录

顺手再演示一个:

import os
from contextlib import contextmanager


@contextmanager
def chdir(path):
    old = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old)


before = os.getcwd()
with chdir('/tmp'):
    print('在 with 里:', os.getcwd())

print('退出后回到:', os.getcwd() == before)

输出(路径以实际为准):

在 with 里: /tmp
退出后回到: True

补一句:Python 3.11 在标准库里加了 contextlib.chdir ,效果跟我们手写的这个一样,可以直接 from contextlib import chdir 。如果你已经在 3.11+ ,就不用自己造轮子了。

ExitStack:动态数量的上下文

到目前为止,我们一直假设「我有 N 个上下文,N 是写代码的时候就知道的」。但是有时候 N 是运行时才知道的——比如「打开用户给的一组文件,全部用完之后再关」。

with 一行写不下,因为名字数量不定;用嵌套 with 写不出来,因为缩进不知道嵌几层。这种时候 ExitStack 就派上用场了。

from contextlib import ExitStack


def make_files(paths):
    with ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        total = sum(len(f.read()) for f in files)
        print(f'共读取 {len(files)} 个文件,{total} 字节')


with open('/tmp/two_drops_a.txt', 'w') as f:
    f.write('a' * 100)
with open('/tmp/two_drops_b.txt', 'w') as f:
    f.write('bb' * 50)
with open('/tmp/two_drops_c.txt', 'w') as f:
    f.write('ccc' * 30)

make_files(['/tmp/two_drops_a.txt', '/tmp/two_drops_b.txt', '/tmp/two_drops_c.txt'])

输出:

共读取 3 个文件,290 字节

stack.enter_context(cm) 把一个上下文管理器「压入栈」,效果就跟把它放在最外层 with 一样。当 with ExitStack() 退出的时候,栈里所有的上下文会按照「后进先出」的顺序被一一退出,跟普通嵌套 with 的语义完全一致。

ExitStack 还有一个特别好用的方法叫 callback ,可以注册一个普通的清理函数,不需要它是上下文管理器:

from contextlib import ExitStack


def cleanup(name):
    print(f'清理 {name}')


with ExitStack() as stack:
    stack.callback(cleanup, 'A')
    stack.callback(cleanup, 'B')
    stack.callback(cleanup, 'C')
    print('正在 with 块内部')

输出:

正在 with 块内部
清理 C
清理 B
清理 A

清理顺序是 C → B → A ,倒着来。这跟函数局部变量析构的顺序一致,符合「先打开后关」的直觉。

ExitStack 还有一些更高级的用法,比如 pop_all 把栈转移到另一个 ExitStack 里,用来实现「事务性清理」——但那是进阶话题了,等各位真正遇到再去翻文档不迟。

补一个比较实用的模式:把多个文件按动态数量打开,万一中途有一个开失败,已经打开的全部要被关掉。ExitStack 天然就帮我们做到了这一点:

from contextlib import ExitStack


def open_safely(paths):
    with ExitStack() as stack:
        files = []
        for p in paths:
            files.append(stack.enter_context(open(p)))
        return [f.name for f in files]


# 准备三个真实存在的文件
for name in ['/tmp/two_drops_x.txt', '/tmp/two_drops_y.txt', '/tmp/two_drops_z.txt']:
    with open(name, 'w') as f:
        f.write('hi')

names = open_safely(['/tmp/two_drops_x.txt', '/tmp/two_drops_y.txt', '/tmp/two_drops_z.txt'])
print(f'安全地处理了 {len(names)} 个文件')

try:
    open_safely(['/tmp/two_drops_x.txt', '/tmp/this_does_not_exist_99999.txt'])
except FileNotFoundError as e:
    print(f'第二个文件失败,但是第一个已经被自动关闭了:{e.filename}')

输出:

安全地处理了 3 个文件
第二个文件失败,但是第一个已经被自动关闭了:/tmp/this_does_not_exist_99999.txt

这就是 ExitStack 在工程里最值钱的地方:「批量打开,一旦失败统一回滚」。如果让我们手写 try/finally 来实现,代码会变得非常啰嗦,而且很容易漏掉某条异常路径。

async with:异步世界的 with

最后简单提一句异步版本。Python 3.5 引入了 async/await 语法,3.7 之后逐渐成为主流。在 async def 函数里,如果要用上下文管理器,需要写成 async with

# 这是一个示意,需要在 async 函数里运行
# async with httpx.AsyncClient() as client:
#     resp = await client.get('https://example.com')

对应的协议方法是 __aenter____aexit__contextlib 也提供了 @asynccontextmanager 装饰器。这部分内容会在 async/await 章节深入讲,这里只是让童鞋们知道有这么一回事,不要在 async 代码里看到 async with 一脸懵。

小实战:写一个 timer 上下文

把这一章学到的东西串起来,我们写一个小工具:进入 with 时记录时间,退出时打印「这段代码耗时多少毫秒」。

先用类的写法:

import time


class Timer:
    def __init__(self, name='匿名代码块'):
        self.name = name

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, tb):
        cost_ms = (time.perf_counter() - self.start) * 1000
        print(f'{self.name} 耗时 {cost_ms:.2f} ms')


with Timer('两点水的睡觉时间'):
    time.sleep(0.05)
    s = sum(range(10000))

输出大约是:

两点水的睡觉时间 耗时 50.46 ms

再用 @contextmanager 重写,对比一下:

import time
from contextlib import contextmanager


@contextmanager
def timer(name='匿名代码块'):
    start = time.perf_counter()
    try:
        yield
    finally:
        cost_ms = (time.perf_counter() - start) * 1000
        print(f'{name} 耗时 {cost_ms:.2f} ms')


with timer('打卡用时'):
    time.sleep(0.03)
    sum(range(5000))

输出大约是:

打卡用时 耗时 30.21 ms

各位选哪个版本?我个人偏爱 @contextmanager 版本,更紧凑。但是当上下文管理器需要保存复杂状态、或者要被复用、或者需要继承的时候,类的写法更合适。两种写法的能力是等价的,看场合选一个就好。

再加点料——如果想让 Timer 既能当上下文管理器,又能当装饰器呢?这两个需求经常一起出现。contextlib 早就替我们想好了,ContextDecorator 这个基类一行代码搞定:

import time
from contextlib import ContextDecorator


class Timer(ContextDecorator):
    def __init__(self, name='匿名代码块'):
        self.name = name

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, tb):
        cost_ms = (time.perf_counter() - self.start) * 1000
        print(f'{self.name} 耗时 {cost_ms:.2f} ms')


@Timer('两点水的函数')
def work():
    time.sleep(0.02)


work()

输出大约是:

两点水的函数 耗时 20.34 ms

而且如果有了 ContextDecorator@contextmanager 装饰器写出来的函数也天然能当装饰器用,不需要再做额外处理:

import time
from contextlib import contextmanager


@contextmanager
def timer(name='匿名代码块'):
    start = time.perf_counter()
    try:
        yield
    finally:
        cost_ms = (time.perf_counter() - start) * 1000
        print(f'{name} 耗时 {cost_ms:.2f} ms')


@timer('打卡函数')
def punch():
    time.sleep(0.01)


punch()

输出大约是:

打卡函数 耗时 10.32 ms

是不是发现 @contextmanager 自带了 ContextDecorator 的能力?所以一个函数同时是装饰器和上下文管理器,毫无违和感。这也是为什么很多日志、性能监控、tracing 库都喜欢用 @contextmanager 写它们的核心 API :一份实现,两种用法。

一个常见的反模式

最后唠叨几句新手最容易踩的几个坑:

反模式一:在 with 外面访问 with 的变量

with open('/tmp/two_drops_d.txt', 'w') as f:
    f.write('hi')

print(f.closed)

输出:

True

f 这个变量在 with 退出后还在,但它指向的文件对象已经被 close 了。这不是错误,但是非常具有误导性——很多新手以为 with 出来后 f 会变成 None 或者直接 NameError ,实则不会。能拿到这个变量但是它已经废了,这是 Python 上下文管理器的设计取舍。

记住:with 块退出后,as 后面的变量仍然存在于当前作用域,但里面的资源已经被释放,碰它大概率会出错。

反模式二:忘了 try/finally

@contextmanager 时如果不写 try/finally ,看起来正常路径下没问题,但是异常一来就资源泄露:

from contextlib import contextmanager


@contextmanager
def bad():
    print('打开资源')
    yield
    print('关闭资源')


try:
    with bad():
        raise ValueError('故意抛个异常')
except ValueError:
    print('外层捕获到了异常')

输出:

打开资源
外层捕获到了异常

「关闭资源」这句话没有被打印,因为异常把 yield 之后的代码全跳过了。所以前面才那么强调,清理逻辑必须放在 finally 里。

反模式三:在 @contextmanager 函数里 yield 多次

from contextlib import contextmanager


@contextmanager
def double_yield():
    print('准备')
    yield 1
    yield 2
    print('结束')


try:
    with double_yield() as v:
        print(f'拿到了 {v}')
except RuntimeError as e:
    print(f'报错了:{e}')

输出:

准备
拿到了 1
报错了:generator didn't stop

@contextmanager 装饰的生成器只能 yield 一次。yield 之前是 __enter__ ,yield 之后是 __exit__ 。如果 yield 了两次,Python 在第二次 yield 的时候发现「这个生成器没有按预期在 yield 之后立即结束」,就会抛 RuntimeError 。这是新手最容易写出来的 bug ,因为生成器函数本身是允许多次 yield 的,但是放到 @contextmanager 这个上下文里就只能 yield 一次。

反模式四:滥用 return True 吞异常

class SwallowEverything:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, tb):
        return True


with SwallowEverything():
    1 / 0
    raise RuntimeError('what?')

print('我居然还在执行')

输出:

我居然还在执行

这种「全吞」式上下文管理器是巨型反模式。它会把真正的 bug 也一起吃掉,让程序在错误状态下继续运行,调试时根本没法定位。return True 应该是非常窄、非常明确的「我就是要忽略这一种异常」的精确动作,而不是「让世界更安静一点」的偷懒手段。

一个综合的小例子:可重用的「水哥打卡审计器」

为了让前面的知识点串起来,我们写一个稍微复杂一点的例子收尾。需求是:每次员工进入工位都要记录开始时间,离开时记录结束时间,期间如果代码出了异常,要把异常类型一并记录。最后再加一个总结报告。

import time
from contextlib import contextmanager


class Auditor:
    def __init__(self):
        self.events = []

    @contextmanager
    def shift(self, name):
        start = time.perf_counter()
        self.events.append({'name': name, 'phase': 'enter'})
        try:
            yield
        except Exception as e:
            self.events.append({
                'name': name,
                'phase': 'error',
                'exception': type(e).__name__,
            })
            raise
        finally:
            cost_ms = (time.perf_counter() - start) * 1000
            self.events.append({
                'name': name,
                'phase': 'leave',
                'cost_ms': round(cost_ms, 2),
            })

    def report(self):
        print(f'共记录 {len(self.events)} 条事件:')
        for e in self.events:
            print(f'  - {e}')


auditor = Auditor()

with auditor.shift('两点水'):
    time.sleep(0.01)

try:
    with auditor.shift('童鞋甲'):
        raise ValueError('代码炸了')
except ValueError:
    pass

with auditor.shift('童鞋乙'):
    time.sleep(0.005)

auditor.report()

输出大约是:

共记录 7 条事件:
  - {'name': '两点水', 'phase': 'enter'}
  - {'name': '两点水', 'phase': 'leave', 'cost_ms': 10.42}
  - {'name': '童鞋甲', 'phase': 'enter'}
  - {'name': '童鞋甲', 'phase': 'error', 'exception': 'ValueError'}
  - {'name': '童鞋甲', 'phase': 'leave', 'cost_ms': 0.16}
  - {'name': '童鞋乙', 'phase': 'enter'}
  - {'name': '童鞋乙', 'phase': 'leave', 'cost_ms': 5.21}

这段代码做对了几件重要的事,各位可以挨个对照一下:

  1. @contextmanager 装饰的是类的方法——这完全合法,方法的 self 在生成器内部依然可用
  2. try/except/finally 三件套俱全:异常路径走 except ,所有路径都走 finally
  3. except Exception as e: ...; raise :先记录,再原样重新抛出,不吞异常
  4. 报告里能看到一次错误事件的完整生命周期:enter → error → leave

这就是上下文管理器在工程中的典型用法——一个对象、几行代码,把「时间统计、异常记录、资源清理」三件事打包到一起。在做性能监控、APM 接入、tracing 时,几乎所有库的核心 API 都长这个样。

小结

with 不是一个孤立的语法糖,它背后是「上下文管理协议」:

  • __enter__ 进入,__exit__ 离开
  • __exit__ 收到三个异常相关参数,返回 True 可以吞掉异常(请慎用)
  • 多个上下文可以用逗号分隔,3.10+ 还能用括号包起来,多行书写
  • contextlib.contextmanager 让我们用一个 yield 替代两个 dunder 方法,但是必须用 try/finally 保证清理代码被执行
  • closingsuppressredirect_stdoutnullcontext 这些小工具能省掉不少样板
  • ExitStack 处理「动态数量的上下文」
  • async with 是异步版本,思路一致,等到 async/await 章节再细聊

凡是有「成对操作」、「资源借还」、「临时切换」、「进入/离开」这种语义的地方,都可以用上下文管理器去封装。一旦养成这个习惯,你写出来的代码不仅短,而且对异常路径天然友好——这才是 with 真正值钱的地方。

下一章我们要深入到 async/await 的世界,去看看 Python 是怎么用一个 async 关键字撬动整个异步生态的。各位先把 with 玩熟,到时候看 async with 会觉得就是顺水推舟的事。