Everyone learns with open(...) and stops. That is a shame, because the protocol behind it is two methods and it solves a whole category of bug: cleanup that does not happen on the paths you forgot about.
The value is that with expands into a try, so the exit runs whether the block finished normally, raised, returned, or broke out of a loop.
The class-based protocol
Two methods. __enter__ returns whatever as binds; __exit__ receives the exception triple, or three Nones on a clean exit.
import timeimport logging
log = logging.getLogger(__name__)
class Timed: """Log how long a block took, whether or not it succeeded."""
def __init__(self, label: str): self.label = label self.elapsed: float | None = None
def __enter__(self) -> "Timed": self._start = time.perf_counter() return self # so the caller can read .elapsed afterwards
def __exit__(self, exc_type, exc, tb) -> None: self.elapsed = time.perf_counter() - self._start outcome = "failed" if exc_type else "ok" log.info("%s %s in %.3fs", self.label, outcome, self.elapsed) # Returning None (falsy) lets any exception propagate. This is what you # want almost always. Returning True would silently swallow it.
with Timed("reindex") as t: rebuild_index()print(f"took {t.elapsed:.3f}s")Careful here
__exit__ returning a truthy value suppresses the exception. This is easy to do by accident, for example ending the method with return self.cleanup() where cleanup() happens to return something truthy. The symptom is errors vanishing with no traceback, which is a miserable thing to debug. Return None explicitly unless suppression is the actual intent.
The generator form, which you will use more often
For simple setup and teardown, @contextmanager is shorter and the exception semantics are hard to get wrong because finally says exactly what you mean.
from contextlib import contextmanager
@contextmanagerdef transaction(conn): """Commit on success, roll back on any exception, always.""" cur = conn.cursor() try: yield cur # everything before this is __enter__ conn.commit() # only reached if the body did not raise except Exception: conn.rollback() raise # re-raise: the caller still needs to know finally: cur.close() # runs on every path
with transaction(conn) as cur: cur.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (100, 1)) cur.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (100, 2))# committed here, or rolled back if either statement raisedTip
The try is not optional. Without it, an exception in the body propagates out of the yield and the code after it never runs, so your teardown is skipped exactly when you need it most. If you write @contextmanager and there is no try/finally, it is almost certainly wrong.
A file lock, which is where this really earns its keep
import fcntlimport osfrom contextlib import contextmanagerfrom pathlib import Path
@contextmanagerdef file_lock(path: Path, *, blocking: bool = True): """Cooperative advisory lock. Releases even if the body raises.""" path.parent.mkdir(parents=True, exist_ok=True) fh = open(path, "w") flags = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB try: fcntl.flock(fh, flags) fh.write(str(os.getpid())) fh.flush() yield except BlockingIOError as err: raise RuntimeError(f"already locked: {path}") from err finally: # Order matters: release, then close. Closing releases it anyway, but # being explicit means the intent survives a later edit. fcntl.flock(fh, fcntl.LOCK_UN) fh.close()
with file_lock(Path("/tmp/import.lock"), blocking=False): run_import() # a second process gets RuntimeError instead of racingThis is the pattern that turns “we sometimes run the import twice and corrupt the table” into a startup error.
Composing them
Nested managers tear down in reverse order, which matters when resources depend on each other:
# B is entered after A, and exited BEFORE A.with database() as db, transaction(db) as tx: tx.execute(...)When the number of resources is not known until runtime, ExitStack is the tool:
from contextlib import ExitStack
def merge(paths: list[str], out: str) -> None: with ExitStack() as stack: # If open() raises on the fourth file, the first three still close. handles = [stack.enter_context(open(p)) for p in paths] with open(out, "w") as sink: for line in roundrobin(*handles): sink.write(line)ExitStack also lets you register plain cleanup callables and, usefully, decide not to clean up:
with ExitStack() as stack: tmp = stack.enter_context(tempfile.TemporaryDirectory()) build_artifacts(tmp) if keep_for_debugging: stack.pop_all() # disarm every cleanup: the directory survivesUseful ones already in the standard library
Reach for these before writing your own:
| Manager | What it does |
|---|---|
open() | closes the file |
tempfile.TemporaryDirectory() | removes the tree on exit |
contextlib.suppress(FileNotFoundError) | replaces a try/except/pass |
contextlib.redirect_stdout(buf) | captures prints from code you cannot change |
contextlib.closing(thing) | adapts an object that only has .close() |
threading.Lock() | acquires and releases |
unittest.mock.patch(...) | patches, then restores |
decimal.localcontext() | scoped precision changes |
suppress in particular is worth the habit:
from contextlib import suppress
# instead of try/except OSError: passwith suppress(FileNotFoundError): Path("stale.lock").unlink()The trap that actually causes incidents: scope
Everything above is about what to wrap. The mistake I see most in production is wrapping the right thing at the wrong scope.
with binds a lifetime to a block. That is exactly right for a unit of work (a transaction, a lock, a temporary directory) and exactly wrong for something expensive you want to keep across calls (a database connection, an HTTP session, a boto3 client).
# Module scope: created once per container, reused across warm invocations._session = boto3.Session()_s3 = _session.client("s3")_db = create_engine(DSN, pool_pre_ping=True) # pre_ping: the container freezes
def handler(event, context): # Per-invocation unit of work, correctly scoped. with _db.begin() as conn: conn.execute(...)Note
pool_pre_ping=True matters on Lambda specifically. The execution environment is frozen between invocations, and the database may close an idle connection while it is frozen. Without a liveness check, the next invocation picks up a dead socket and fails once before recovering.
Async, briefly
The async protocol mirrors the sync one with __aenter__/__aexit__, and contextlib.asynccontextmanager for the generator form:
from contextlib import asynccontextmanager
@asynccontextmanagerasync def http_session(): session = aiohttp.ClientSession() try: yield session finally: await session.close() # await in teardown is the whole reason this exists
async def fetch_all(urls): async with http_session() as s: return await asyncio.gather(*(s.get(u) for u in urls))Note async with, not with. Using a plain with on an async manager raises AttributeError about __enter__, which is a confusing error for what is really a one-keyword mistake.
