Python3 with
with 语句的语法、上下文管理器协议、典型应用场景与自定义实现。
为什么需要 with 语句?
Files, locks and connections must be released in pairs. try/finally guarantees it but is boilerplate-heavy and easy to forget. with compresses acquire → use → release into one block.
f = open('demo.txt', 'w', encoding='utf-8')
try:
f.write("hello")
finally:
f.close()
print("closed:", f.closed)
with open('demo.txt', 'w', encoding='utf-8') as g:
g.write("hello")
print("closed:", g.closed)closed: True
closed: Truewith 语句的基本语法
with expr as var: — the expression yields a context manager and as binds whatever __enter__ returns. Several resources may be managed at once, comma-separated.
with open('a.txt', 'w', encoding='utf-8') as a, \
open('b.txt', 'w', encoding='utf-8') as b:
a.write("A")
b.write("B")
print(open('a.txt', encoding='utf-8').read())
print(open('b.txt', encoding='utf-8').read())
print(a.closed, b.closed)A
B
True Truewith 语句的工作原理
Entering calls __enter__; leaving calls __exit__(exc_type, exc, tb). If the block raised, returning True from __exit__ suppresses the exception; False or None lets it propagate.
class Trace:
def __enter__(self):
print("enter")
return self
def __exit__(self, et, e, tb):
print("exit", et.__name__ if et else None)
return False
with Trace():
print("body")
try:
with Trace():
raise ValueError("boom")
except ValueError:
print("propagated")
class Swallow(Trace):
def __exit__(self, et, e, tb):
print("swallowed", et.__name__)
return True
with Swallow():
raise ValueError("hidden")
print("after")enter
body
exit None
enter
exit ValueError
propagated
enter
swallowed ValueError
after实际应用场景
The three common cases: file I/O, thread/process locks, and database transactions or connections. All share the need for paired release even on exceptions.
import threading
import tempfile
import os
lock = threading.Lock()
with lock:
print("lock held:", lock.locked())
print("lock released:", lock.locked())
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, 'x.txt')
with open(path, 'w', encoding='utf-8') as f:
f.write("data")
print("wrote:", os.path.exists(path))
print("dir removed:", os.path.exists(d))lock held: True
lock released: False
wrote: True
dir removed: False创建自定义上下文管理器
Two ways: a class with __enter__/__exit__, or decorate a generator with contextlib.contextmanager — the latter is shorter: code before yield is setup, after it is teardown.
from contextlib import contextmanager
import time
@contextmanager
def timer(label):
start = time.perf_counter()
try:
yield label
finally:
print(label, "took > 0s:", time.perf_counter() - start >= 0)
with timer("work") as name:
total = sum(range(1000))
print(name, total)work took > 0s: True
work 499500为什么需要 with 语句?
打开文件、加锁、开连接都必须成对释放。手写 try/finally 能保证,但样板代码多、容易漏。with 把「获取 → 使用 → 释放」压缩成一个块。
f = open('demo.txt', 'w', encoding='utf-8')
try:
f.write("hello")
finally:
f.close()
print("closed:", f.closed)
with open('demo.txt', 'w', encoding='utf-8') as g:
g.write("hello")
print("closed:", g.closed)closed: True
closed: Truewith 语句的基本语法
with 表达式 as 变量: —— 表达式返回上下文管理器,as 绑定 __enter__ 的返回值。可以一次管多个资源,用逗号分隔。
with open('a.txt', 'w', encoding='utf-8') as a, \
open('b.txt', 'w', encoding='utf-8') as b:
a.write("A")
b.write("B")
print(open('a.txt', encoding='utf-8').read())
print(open('b.txt', encoding='utf-8').read())
print(a.closed, b.closed)A
B
True Truewith 语句的工作原理
进入 with 时调用 __enter__,退出时调用 __exit__(exc_type, exc, tb)。若块内抛异常,__exit__ 返回 True 会吞掉异常,返回 False(或 None)则继续抛出。
class Trace:
def __enter__(self):
print("enter")
return self
def __exit__(self, et, e, tb):
print("exit", et.__name__ if et else None)
return False
with Trace():
print("body")
try:
with Trace():
raise ValueError("boom")
except ValueError:
print("propagated")
class Swallow(Trace):
def __exit__(self, et, e, tb):
print("swallowed", et.__name__)
return True
with Swallow():
raise ValueError("hidden")
print("after")enter
body
exit None
enter
exit ValueError
propagated
enter
swallowed ValueError
after实际应用场景
最常见的三类:文件读写、线程/进程锁、数据库事务与连接。共同点是「必须配对释放,且异常时也不能漏」。
import threading
import tempfile
import os
lock = threading.Lock()
with lock:
print("lock held:", lock.locked())
print("lock released:", lock.locked())
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, 'x.txt')
with open(path, 'w', encoding='utf-8') as f:
f.write("data")
print("wrote:", os.path.exists(path))
print("dir removed:", os.path.exists(d))lock held: True
lock released: False
wrote: True
dir removed: False创建自定义上下文管理器
两种写法:类实现 __enter__/__exit__,或用 contextlib.contextmanager 装饰一个生成器函数 —— 后者更短,yield 之前是进入逻辑,之后是清理逻辑。
from contextlib import contextmanager
import time
@contextmanager
def timer(label):
start = time.perf_counter()
try:
yield label
finally:
print(label, "took > 0s:", time.perf_counter() - start >= 0)
with timer("work") as name:
total = sum(range(1000))
print(name, total)work took > 0s: True
work 499500