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

HelperUse
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())
ExitStackdynamic or conditional number of managers
AsyncExitStacksame, 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