Python Cheatsheet
Context Managers
Use this Python reference while you build software engineering projects, review code, or refresh the syntax you reach for most.
What with Guarantees
with expr as name: calls expr.__enter__(), binds its return value to name,
and calls __exit__ on the way out. __exit__ runs whether the block finished,
raised, returned, breaked, or continued. It does not run if the process is
killed or os._exit is called.
with open("data.txt", encoding="utf-8") as f: # closed even if this raises process(f.read()) # Desugared f = open("data.txt", encoding="utf-8") try: process(f.read()) finally: f.close()
as name binds whatever __enter__ returns, which is not always the manager.
open() returns the file, but threading.Lock().__enter__() returns True, so
with lock as x gives you a useless x. Just write with lock:.
Multiple Managers
Commas nest left to right, so the first one entered is the last one exited. From 3.10 you can wrap the list in parentheses and break lines.
with open("in.txt") as src, open("out.txt", "w") as dst: dst.write(src.read()) # Parenthesized, trailing comma allowed (3.10+) with ( open("in.txt", encoding="utf-8") as src, open("out.txt", "w", encoding="utf-8") as dst, lock, ): dst.write(src.read().upper()) # Nested form is identical in behavior with open("in.txt") as src: with open("out.txt", "w") as dst: ...
Writing __enter__ and __exit__
__exit__(self, exc_type, exc_val, exc_tb) gets (None, None, None) on a clean
exit. Returning a truthy value swallows the exception, returning None or
False lets it propagate. Only suppress deliberately.
class Transaction: def __init__(self, conn): self.conn = conn def __enter__(self): self.conn.execute("BEGIN") return self.conn # this is what `as` binds def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: self.conn.execute("COMMIT") else: self.conn.execute("ROLLBACK") return False # re-raise the original error class Swallow: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): return isinstance(exc_val, ValueError) # True hides ValueError only with Swallow(): raise ValueError("boom") print("still running") # reached, the error was suppressed
Subclass contextlib.AbstractContextManager to inherit a default __enter__
that returns self, and to satisfy isinstance checks. Files already register,
issubclass(io.IOBase, AbstractContextManager) is True.
@contextmanager
The shortest way to write one: everything before yield is setup, the yielded
value is what as binds, and cleanup goes in a finally so it also runs on an
exception.
from contextlib import contextmanager import time @contextmanager def timed(label): start = time.perf_counter() try: yield # yield nothing if there is no handle finally: print(f"{label}: {time.perf_counter() - start:.3f}s") with timed("query"): run_query() @contextmanager def temp_attr(obj, name, value): old = getattr(obj, name) setattr(obj, name, value) try: yield obj finally: setattr(obj, name, old)
Two rules that bite. Without the try/finally, an exception in the body skips
your cleanup entirely. And the object returned by the decorated function is
single-use, entering the same object twice raises, so call the factory again
rather than storing one instance. A @contextmanager function also doubles as a
decorator: @timed("job") on a function times each call.
contextlib Toolkit
| Helper | Use |
|---|---|
closing(thing) | wraps anything with .close() but no with support |
suppress(*excs) | ignore listed exceptions, replaces empty except: pass |
redirect_stdout(f) | capture or divert print output |
redirect_stderr(f) | same for sys.stderr |
nullcontext(x) | a do-nothing manager, for optional managers |
chdir(path) (3.11+) | change cwd and restore it, not thread safe |
aclosing(thing) | closing for async generators (aclose()) |
ExitStack | dynamic or conditional number of managers |
AsyncExitStack | same, for async with |
from contextlib import closing, suppress, redirect_stdout, nullcontext from urllib.request import urlopen import io, os with closing(urlopen("https://example.com")) as resp: body = resp.read() with suppress(FileNotFoundError): os.remove("maybe-missing.tmp") # no-op if absent buf = io.StringIO() with redirect_stdout(buf): print("captured") buf.getvalue() # 'captured\n' def run(lock=None, verbose=False): with (lock or nullcontext()): # no branching on lock is None ...
suppress and redirect_stdout are reentrant, so the same object can be nested.
A file object and a @contextmanager object are not.
ExitStack
For a count of managers you do not know until runtime, or for cleanup registered mid-block. Everything unwinds in reverse order.
from contextlib import ExitStack paths = ["a.txt", "b.txt", "c.txt"] with ExitStack() as stack: files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths] stack.callback(print, "runs first on unwind") merged = "".join(f.read() for f in files) # every file is closed here # Conditional entry with ExitStack() as stack: if use_lock: stack.enter_context(lock) conn = stack.enter_context(connect()) # Commit-or-rollback idiom: cancel the cleanup once you know you succeeded with ExitStack() as stack: stack.callback(os.remove, "partial.tmp") write_everything("partial.tmp") os.replace("partial.tmp", "final.txt") stack.pop_all() # success, skip the delete
Async Context Managers
async with calls __aenter__ and __aexit__, both coroutines. Use it for
anything whose setup or teardown awaits, such as a connection pool or an HTTP
session.
import asyncio from contextlib import asynccontextmanager, AsyncExitStack class Pool: async def __aenter__(self): self.conn = await connect() return self.conn async def __aexit__(self, exc_type, exc_val, exc_tb): await self.conn.close() return False @asynccontextmanager async def session(url): client = await open_client(url) try: yield client finally: await client.aclose() async def main(): async with session("https://api.example.com") as client: data = await client.get("/items") async with AsyncExitStack() as stack: clients = [await stack.enter_async_context(session(u)) for u in urls] asyncio.run(main())
asyncio.Lock, asyncio.Semaphore, asyncio.timeout (3.11+), and httpx
clients are all async managers. A plain with on them raises TypeError.
Managers People Actually Use
import threading, tempfile, sqlite3, os from pathlib import Path from decimal import localcontext, Decimal # Locks (never leaks on an exception, unlike manual acquire/release) lock = threading.Lock() with lock: shared_counter += 1 # RLock is reentrant, Lock deadlocks if the same thread re-acquires it with threading.RLock() as _: pass # Temp file and temp dir, both cleaned up on exit with tempfile.TemporaryDirectory() as d: Path(d, "scratch.json").write_text("{}") with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=True) as tmp: tmp.write("a,b\n") # sqlite3 connection as a manager commits or rolls back the TRANSACTION # (it does NOT close the connection, close it yourself) conn = sqlite3.connect(":memory:") with conn: conn.execute("CREATE TABLE t (x int)") conn.close() # Decimal precision, scoped with localcontext() as ctx: ctx.prec = 50 Decimal(1) / Decimal(7) # unittest.mock.patch, pytest.raises, and warnings.catch_warnings are managers too