---
title: "Python Context Managers: Writing Clean Resource Handling with `with` Statements"
description: "A Python snippet on writing custom context managers with both the class-based `__enter__`/`__exit__` protocol and the `@contextmanager` decorator. Covers database transactions, file locks, timers, temporary directories, ExitStack for dynamic composition, and the __exit__ return value that silently swallows exceptions."
author: "Mohammad Abu Mattar"
canonical: https://mkabumattar.com/codesnippets/post/python-context-managers-resource-handling
---

# Python Context Managers: Writing Clean Resource Handling with `with` Statements

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.

_What `with` expands to, and the two ways to write one_

## The class-based protocol

Two methods. `__enter__` returns whatever `as` binds; `__exit__` receives the exception triple, or three `None`s on a clean exit.

```python

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")
```

`__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.

```python
from contextlib import contextmanager

@contextmanager
def 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 raised
```

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

```python

from contextlib import contextmanager
from pathlib import Path

@contextmanager
def 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 racing
```

This 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:

```python
# 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:

```python
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:

```python
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 survives
```

## Useful 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:

```python
from contextlib import suppress

# instead of try/except OSError: pass
with 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.

_Right pattern, wrong scope, on AWS Lambda_

`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).

```python
# 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(...)
```

`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:

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async 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.

## References

- [`with` statement reference (Python docs)](https://docs.python.org/3/reference/compound_stmts.html#the-with-statement)
- [With statement context managers (data model)](https://docs.python.org/3/reference/datamodel.html#with-statement-context-managers)
- [`contextlib` module](https://docs.python.org/3/library/contextlib.html)
- [`contextlib.ExitStack`](https://docs.python.org/3/library/contextlib.html#contextlib.ExitStack)
- [`contextlib.asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager)
- [PEP 343, the "with" statement](https://peps.python.org/pep-0343/)
- [SQLAlchemy connection pooling and `pool_pre_ping`](https://docs.sqlalchemy.org/en/20/core/pooling.html#disconnect-handling-pessimistic)
- [Using Amazon RDS Proxy with Lambda](https://docs.aws.amazon.com/lambda/latest/dg/services-rds-tutorial.html)