A context manager pairs setup with guaranteed teardown. with runs the teardown however the block exits — normally, by return, or by an exception.
The protocol
class Resource:
def __init__(self, name):
self.name = name
def __enter__(self):
print(f"open {self.name}")
return self # bound to the "as" target
def __exit__(self, exc_type, exc, tb):
print(f"close {self.name} (error: {exc_type.__name__ if exc_type else None})")
return False # False/None → let the exception propagate
with Resource("db") as r:
print("using", r.name)
try:
with Resource("file"):
raise KeyError("boom")
except KeyError:
print("caught outside")open db
using db
close db (error: None)
open file
close file (error: KeyError)
caught outsideEvaluate
The expression after
withproduces the manager.Enter
__enter__()runs; its return value is bound to theasname.Body
The indented block runs.
Exit
__exit__(exc_type, exc, tb)always runs — all three areNoneif there was no exception.Suppress?
If
__exit__returns a truthy value, the exception is swallowed. Almost always returnFalse.
A reusable class-based manager
import time
class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, *exc):
self.elapsed = time.perf_counter() - self.start
return False
with Timer() as t:
sum(range(100_000))
print(t.elapsed > 0) # True@contextmanager: a generator becomes a manager
Everything before yield is __enter__, the yielded value is the as target, everything after is __exit__. Wrap the yield in try/finally or the cleanup is skipped when the body raises.
from contextlib import contextmanager
@contextmanager
def temporary_setting(config, key, value):
old = config[key]
config[key] = value
try:
yield config
finally:
config[key] = old # restored even if the body raises
cfg = {"debug": False}
with temporary_setting(cfg, "debug", True):
print(cfg["debug"]) # True
print(cfg["debug"]) # Falsecontextlib helpers
import io
from contextlib import suppress, redirect_stdout, nullcontext, closing
with suppress(KeyError): # ignore one specific exception
{}["missing"]
buf = io.StringIO()
with redirect_stdout(buf): # capture print output
print("hidden")
print(buf.getvalue().strip()) # hidden
def process(lock=None):
with lock or nullcontext(): # "maybe a lock" without an if/else
return "processed"
print(process()) # processed
class Conn:
def close(self):
self.closed = True
with closing(Conn()) as c: # calls c.close() on exit
pass
print(c.closed) # True| Helper | Purpose |
|---|---|
suppress(*excs) | Ignore the listed exceptions inside the block |
redirect_stdout(f) / redirect_stderr(f) | Temporarily send prints somewhere else |
closing(obj) | Call obj.close() on exit, for objects without __exit__ |
nullcontext(value) | A do-nothing manager — for optional context |
ExitStack() | Enter a dynamic number of managers; unwinds them in reverse |
chdir(path) | Temporarily change the working directory (3.11+) |
ContextDecorator | A manager that can also be used as @decorator |
asynccontextmanager | The async with version of @contextmanager |
ExitStack: many resources at once
import tempfile
from pathlib import Path
from contextlib import ExitStack
folder = Path(tempfile.mkdtemp())
names = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
files = [stack.enter_context(open(folder / n, "w", encoding="utf-8")) for n in names]
for i, f in enumerate(files):
f.write(str(i))
# every file is closed here, even if one open() had failed half-way
print(all(f.closed for f in files), (folder / "c.txt").read_text()) # True 2Several managers in one with
import io
src, dst = io.StringIO("copy me"), io.StringIO()
with (
src as reader, # parenthesised form (3.10+) allows one per line
dst as writer,
):
writer.write(reader.read())
copied = writer.getvalue()
print(copied, src.closed, dst.closed) # copy me True TrueAsync context managers
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def session(name):
await asyncio.sleep(0) # e.g. open a network connection
try:
yield f"session:{name}"
finally:
await asyncio.sleep(0) # e.g. close it
async def main():
async with session("api") as s:
return s
print(asyncio.run(main())) # session:apiWhat happens if an exception occurs inside a with block?
How do you write a context manager without a class?
When would you use ExitStack?