Context managers

Ready-made managers

The standard library has several you'll reach for:

import os
import sqlite3
from contextlib import ExitStack, suppress

with suppress(FileNotFoundError):
    os.remove("does-not-exist.tmp")
print("no error: suppress ignored that one kind of exception")

names = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
    files = [stack.enter_context(open(n, "w")) for n in names]
    for f, n in zip(files, names):
        f.write(n)
print("all closed:", all(f.closed for f in files))

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE stock (sku TEXT PRIMARY KEY, qty INTEGER CHECK (qty >= 0))")
db.execute("INSERT INTO stock VALUES ('TE-101', 3)")
try:
    with db:
        db.execute("UPDATE stock SET qty = qty + 5 WHERE sku = 'TE-101'")
        db.execute("UPDATE stock SET qty = qty - 20 WHERE sku = 'TE-101'")
except sqlite3.IntegrityError as e:
    print("rolled back:", e)
print(db.execute("SELECT qty FROM stock").fetchone())

• suppress(SomeError) ignores one specific exception. Use it only when the error genuinely doesn't matter.
• ExitStack manages any number of context managers, even when you don't know how many in advance. Everything entered is cleaned up when the block ends.
• A sqlite3 connection used with with is a transaction: if the block raises, every change in it is undone (rolled back); if it finishes, all of them are saved (committed). Here the second update would make stock negative, so both updates are undone and the quantity stays 3.

from contextlib import contextmanager

@contextmanager
def override(settings, key, value):
    yield


config = {"currency": "USD"}
with override(config, "currency", "EUR"):
    print(config["currency"])
print(config["currency"])

Where context managers appear

Once you recognize the shape, "set something up, do some work, always clean up", you'll see context managers all over real code:

• Files and network connections: open, requests.Session(), smtplib.SMTP, socket objects.
• Locks in threaded code: with lock: takes the lock and always releases it, so one crash can't freeze every other thread.
• Databases: sessions and transactions in sqlite3 and SQLAlchemy, where the block's changes are saved together or not at all.
• Temporary state: tempfile.TemporaryDirectory() deletes its folder afterwards; decimal.localcontext() changes precision for one calculation; tests use pytest.raises and patching helpers to change something only for the duration of a block.
• Timing and profiling: a timer around a block, as you built above.

A useful test for your own code: whenever you write a try/finally whose finally undoes something the try set up, ask whether it should be a context manager. Naming the pattern once means every caller gets the cleanup right without thinking about it.

Common mistakes

A few slips catch almost everyone the first time:

• Using the resource after the block. with open(...) as f: closes f at the end of the block, so reading from it on the next line fails with "I/O operation on closed file". Do the work inside, or keep the data you read, not the file.
• A @contextmanager generator that yields twice, or never. It must yield exactly once: that one yield is where the block runs. Yielding inside a loop raises "generator didn't stop".
• Forgetting that the block's exception arrives at the yield. Code after a bare yield doesn't run when the block fails. Cleanup belongs in finally.
• Opening many files in a loop without closing them. Opening thousands of files without with can exhaust what the operating system allows. Either close each one in its own with, or hold them in an ExitStack.
• Swallowing errors by accident. A class whose __exit__ returns a true value hides every exception raised in the block, so bugs vanish silently. Return False, or nothing at all, unless hiding one specific error is the whole point.

import os
import tempfile
from contextlib import contextmanager

@contextmanager
def atomic_write(path):
    folder = os.path.dirname(os.path.abspath(path))
    fd, temp_path = tempfile.mkstemp(dir=folder, suffix=".tmp")
    try:
        with os.fdopen(fd, "w") as f:
            yield f
        os.replace(temp_path, path)
    except BaseException:
        os.remove(temp_path)
        raise

with atomic_write("prices.json") as f:
    f.write('{"TE-101": 200}')
print(open("prices.json").read())

try:
    with atomic_write("prices.json") as f:
        f.write('{"TE-101": 17')
        raise RuntimeError("crashed halfway")
except RuntimeError:
    pass
print(open("prices.json").read())
print([n for n in os.listdir(".") if n.endswith(".tmp")])