Context managers beyond with open(): owning resource lifecycles in production Python

Every Python developer knows with open(...) as f. Fewer treat context managers as what they really are: the language's mechanism for owning a lifecycle. Anything with a beginning and an end (a transaction, a request, a connection pool, a batch of concurrent tasks, a temporary override) has a boundary. A context manager makes that boundary explicit, testable and impossible to forget.

In long-running services and integrations, those boundaries are where reliability is won or lost. This post skips the basics and focuses on the patterns I rely on in production code.

1. Error-translation boundaries#

Low-level exceptions should not leak across architectural layers. A TimeoutError from a socket means nothing to the code orchestrating a business process; SystemUnavailable("erp") does. A context manager is the cleanest place to translate one into the other, because it wraps exactly the region where the low-level calls happen.

from contextlib import contextmanager


class IntegrationError(Exception):
    """Base class for failures talking to an external system."""


class SystemUnavailable(IntegrationError):
    pass


@contextmanager
def boundary(system):
    try:
        yield
    except (TimeoutError, ConnectionError) as exc:
        raise SystemUnavailable(f"{system} is unavailable: {exc}") from exc
    except OSError as exc:
        raise IntegrationError(f"{system} failed: {exc}") from exc


with boundary("erp"):
    push_orders(batch)

Callers now handle one small, meaningful exception hierarchy, while from exc keeps the original traceback for debugging. Order matters here: TimeoutError and ConnectionError are subclasses of OSError, so the specific handler must come first.

2. Transactions and the unit of work#

A context manager is the natural shape of a transaction: commit if the block succeeds, roll back if it fails. But be careful with what libraries give you for free.

Owning the whole lifecycle explicitly removes the ambiguity:

import sqlite3
from contextlib import closing, contextmanager


@contextmanager
def unit_of_work(path):
    with closing(sqlite3.connect(path)) as conn:
        try:
            yield conn
        except BaseException:
            conn.rollback()
            raise
        else:
            conn.commit()

closing() guarantees the connection is closed; the try/except/else guarantees exactly one of commit or rollback. Catching BaseException (and always re-raising) means a KeyboardInterrupt or task cancellation also rolls back instead of leaving a half-written transaction.

3. Request-scoped context with contextvars#

Correlating log lines across a whole run (or a whole request) is invaluable, but threading a run_id parameter through every function is noise. contextvars provides context-local state that works correctly across threads and asyncio tasks. A context manager scopes it precisely:

import logging
from contextlib import contextmanager
from contextvars import ContextVar

run_id = ContextVar("run_id", default="-")


@contextmanager
def run_context(value):
    token = run_id.set(value)
    try:
        yield
    finally:
        run_id.reset(token)


class RunIdFilter(logging.Filter):
    def filter(self, record):
        record.run_id = run_id.get()
        return True


handler = logging.StreamHandler()
handler.addFilter(RunIdFilter())
handler.setFormatter(logging.Formatter("%(asctime)s [%(run_id)s] %(name)s %(message)s"))

Every log line emitted inside with run_context("a1b2c3"): now carries the run ID automatically, including lines from libraries you don't control. Using reset(token) rather than setting the old value back restores the previous state correctly, even when contexts are nested.

4. Transferring ownership with ExitStack#

Constructors that acquire several resources have a classic bug: if the third acquisition fails, the first two leak. ExitStack solves it. Its pop_all() method transfers ownership to the object once construction succeeds:

from contextlib import ExitStack


class MultiFileReader:
    def __init__(self, paths):
        with ExitStack() as stack:
            self.files = [stack.enter_context(open(path)) for path in paths]
            # Construction succeeded: take ownership, so nothing closes on exiting this block
            self._close = stack.pop_all().close

    def close(self):
        self._close()

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.close()

If any open() fails, the stack closes every file opened so far and the exception propagates. If they all succeed, responsibility moves to the object. ExitStack.callback() extends the same guarantee to clean-up that isn't a context manager at all, such as stack.callback(shutil.rmtree, workdir).

5. Structured concurrency and deadlines#

In async code, the lifecycle to own is often a group of tasks. Python 3.11 added two context managers that change how concurrent integrations should be written:

import asyncio


async def sync_batches(batches, send, deadline=30):
    async with asyncio.timeout(deadline):
        async with asyncio.TaskGroup() as group:
            for batch in batches:
                group.create_task(send(batch))


try:
    asyncio.run(sync_batches(batches, send))
except* TimeoutError:
    log.error("sync exceeded its deadline")
except* IntegrationError as group:
    for error in group.exceptions:
        log.error("batch failed: %s", error)

TaskGroup guarantees that when the block exits, every task has finished: if one fails, its siblings are cancelled and all failures are raised together as an ExceptionGroup. asyncio.timeout() puts a single deadline over the whole operation. except* then handles each kind of failure separately. No orphaned tasks, no forgotten deadlines.

6. One object, two interfaces: ContextDecorator#

Some lifecycles apply to a block in one place and a whole function in another. Inheriting from ContextDecorator gives both from a single implementation:

import time
from contextlib import ContextDecorator


class timed(ContextDecorator):
    def __init__(self, label, log):
        self.label, self.log = label, log

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, *exc_info):
        self.log.info("%s took %.3fs", self.label, time.perf_counter() - self.start)
        return False


@timed("nightly sync", log)
def nightly_sync(): ...


with timed("mapping", log):
    mapped = [transform(row) for row in rows]

7. Test the unhappy path#

The whole point of a context manager is what happens when things go wrong, so that is what the tests must prove:

import pytest

from app.db import unit_of_work


def test_failed_block_rolls_back(tmp_path):
    path = tmp_path / "test.db"
    with unit_of_work(path) as conn:
        conn.execute("CREATE TABLE orders (number TEXT)")

    with pytest.raises(RuntimeError):
        with unit_of_work(path) as conn:
            conn.execute("INSERT INTO orders VALUES ('PO-1')")
            raise RuntimeError("boom")

    with unit_of_work(path) as conn:
        assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0

Design rules I follow#

1. Every resource has exactly one owner. The owner is visible in the code as a with block or an object with close().
2. Boundaries translate errors, so each layer speaks its own language.
3. Clean-up lives in finally, else or __exit__, never only on the happy path.
4. Never swallow exceptions by accident. Return True from __exit__ only when suppression is the explicit purpose.
5. Re-raise BaseException after rolling back, so cancellation and interrupts still propagate.
6. Prefer structured concurrency. Tasks should never outlive the block that started them.
7. Test the failure paths, because that is the only path the context manager truly exists for.

Treat context managers as a design tool rather than a convenience. Whole classes of production incidents (leaked connections, half-committed data, orphaned tasks, untraceable logs) simply stop happening.