Context Managers and Resource Management

Context Managers and Resource Management

After reading this chapter, you will master Resource Acquisition Is Initialization (RAII) using Python’s with statement, implement the __enter__ and __exit__ dunder protocol, author lightweight context managers using @contextlib.contextmanager, and dynamically manage multi-resource lifecycles with contextlib.ExitStack.

Mental model

In systems programming and backend engineering, resources (file descriptors, database connections, mutex locks, network sockets) must be deterministic and leak-free. Relying on manual cleanup (file.close()) fails whenever an unexpected exception interrupts execution.

The Context Management Protocol binds resource allocation to the entry of a code block and guarantees release upon exit:

with acquire_resource() as target:
        │
        ├─ 1. Calls target = resource.__enter__()
        │
        ├─ 2. Executes with-block body
        │     │
        │     ├─ Success ────────────▶ Calls resource.__exit__(None, None, None)
        │     │
        │     └─ Exception Raised ───▶ Calls resource.__exit__(exc_type, exc_val, exc_tb)
        │                                  │
        │                                  ├─ Returns True  ──▶ Exception is SUPPRESSED
        │                                  └─ Returns False ──▶ Exception PROPAGATES
        ▼
[ Resumes Outer Execution ]

---

Minimal example

Save as transaction_manager.py:

# transaction_manager.py
from typing import Any

class MockDatabaseTransaction:
    def __init__(self, tx_id: str) -> None:
        self.tx_id = tx_id
        self.active = False

    def __enter__(self) -> "MockDatabaseTransaction":
        self.active = True
        print(f"[{self.tx_id}] BEGIN TRANSACTION: Locks acquired.")
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool:
        if exc_type is not None:
            print(f"[{self.tx_id}] ROLLBACK: Transaction aborted due to {exc_type.__name__}: {exc_val}")
            self.active = False
            return False  # Do not suppress exception; propagate to caller
        
        print(f"[{self.tx_id}] COMMIT: Transaction changes committed safely.")
        self.active = False
        return True

def main() -> None:
    # 1. Successful transaction
    print("--- Running successful transaction ---")
    with MockDatabaseTransaction("TX_101") as tx:
        print(f"  Writing record into transaction {tx.tx_id}...")

    # 2. Failed transaction with automatic rollback
    print("\n--- Running failed transaction ---")
    try:
        with MockDatabaseTransaction("TX_102") as tx:
            print(f"  Writing record into transaction {tx.tx_id}...")
            raise RuntimeError("Database constraint violation on unique constraint")
    except RuntimeError as err:
        print(f"  Caller caught error: {err}")

if __name__ == "__main__":
    main()

Run via uv run python transaction_manager.py:

--- Running successful transaction ---
[TX_101] BEGIN TRANSACTION: Locks acquired.
  Writing record into transaction TX_101...
[TX_101] COMMIT: Transaction changes committed safely.

--- Running failed transaction ---
[TX_102] BEGIN TRANSACTION: Locks acquired.
  Writing record into transaction TX_102...
[TX_102] ROLLBACK: Transaction aborted due to RuntimeError: Database constraint violation on unique constraint
  Caller caught error: Database constraint violation on unique constraint

---

Worked examples

Case 1: Benchmark Timer with @contextlib.contextmanager

Writing a full class with __enter__ and __exit__ is often unnecessary for simple scope management. The @contextmanager decorator turns a generator into a clean context manager:

# execution_timer.py
import time
from contextlib import contextmanager
from collections.abc import Generator

@contextmanager
def benchmark(label: str) -> Generator[None, None, None]:
    start_time = time.perf_counter()
    print(f"[{label}] Started execution...")
    try:
        # yield suspends execution and yields control to the with-block
        yield
    finally:
        # Code in finally ALWAYS runs when the with-block terminates
        elapsed = time.perf_counter() - start_time
        print(f"[{label}] Finished in {elapsed * 1000:.2f} ms")

if __name__ == "__main__":
    with benchmark("Data Aggregation"):
        total = sum(x * x for x in range(500_000))
        print(f"  Computed sum: {total}")

Run:

uv run python execution_timer.py

Output:

[Data Aggregation] Started execution...
  Computed sum: 41666541666750000
[Data Aggregation] Finished in 24.85 ms

Case 2: Clean Error Suppression with contextlib.suppress

Instead of writing a verbose try/except: pass block to ignore expected non-fatal errors (such as deleting a file that might not exist), suppress() makes the intent declarative:

# safe_cleanup.py
import os
from contextlib import suppress

def remove_temporary_lock(lock_file_path: str) -> None:
    # Replaces:
    # try:
    #     os.remove(lock_file_path)
    # except FileNotFoundError:
    #     pass
    with suppress(FileNotFoundError):
        os.remove(lock_file_path)
        print(f"Removed lock file: {lock_file_path}")

if __name__ == "__main__":
    # Target does not exist; suppressed cleanly without error
    remove_temporary_lock("/tmp/non_existent_cluster_lock.pid")
    print("Cleanup executed without raising exceptions.")

Run:

uv run python safe_cleanup.py

Output:

Cleanup executed without raising exceptions.

Case 3: Dynamic Multi-Resource Coordination with ExitStack

When the number of files, sockets, or locks is unknown at authoring time (e.g. merging an arbitrary list of configuration files), a static with open(f1), open(f2): statement is impossible. contextlib.ExitStack manages dynamic collections:

# dynamic_file_merger.py
import tempfile
from contextlib import ExitStack
from pathlib import Path

def merge_log_files(source_paths: list[Path], output_path: Path) -> int:
    total_lines = 0
    with ExitStack() as stack:
        # Dynamically register all files into the stack
        input_handles = [stack.enter_context(p.open("r")) for p in source_paths]
        out_handle = stack.enter_context(output_path.open("w"))

        for handle in input_handles:
            for line in handle:
                out_handle.write(line)
                total_lines += 1

    # ALL handles guaranteed closed here, even if an exception was raised
    return total_lines

if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as tmpdir:
        td = Path(tmpdir)
        f1 = td / "app_01.log"
        f2 = td / "app_02.log"
        f1.write_text("2026-09-07 node01 startup\n")
        f2.write_text("2026-09-07 node02 startup\n")

        merged = td / "merged.log"
        count = merge_log_files([f1, f2], merged)
        print(f"Merged {count} lines into {merged.name}:")
        print(merged.read_text().strip())

Run:

uv run python dynamic_file_merger.py

Output:

Merged 2 lines into merged.log:
2026-09-07 node01 startup
2026-09-07 node02 startup

---

Pitfalls

Pitfall 1: Accidentally Suppressing All Exceptions in __exit__

If __exit__ returns any truthy value (like True or a non-empty object), CPython suppresses every exception raised inside the with block:

# THE BUG:
def __exit__(self, exc_type, exc_val, exc_tb):
    self.cleanup()
    return True  # DANGEROUS! Swallows NameError, TypeError, and syntax bugs!

# THE FIX: Only return True if you specifically intended to handle that exact error
def __exit__(self, exc_type, exc_val, exc_tb):
    self.cleanup()
    if exc_type is ResourceUnavailableError:
        return True  # Suppress only this known error
    return False     # Propagate everything else

Pitfall 2: Omitting try/finally in @contextmanager Generators

In a generator context manager, if an exception is raised inside the caller’s with block, it is thrown into the generator at the point of the yield. Without try...finally, code after yield will never execute:

# THE BUG:
@contextmanager
def acquire_lock():
    lock.acquire()
    yield
    lock.release()  # NEVER RUNS if caller raises an exception inside 'with'!

# THE FIX:
@contextmanager
def acquire_lock():
    lock.acquire()
    try:
        yield
    finally:
        lock.release()  # ALWAYS RUNS

---

Exercises

1. Implement a class-based context manager TemporaryDirectoryChanger that switches the current working directory to a target directory in __enter__ and restores the original directory in __exit__.
2. Write a generator context manager override_env(key: str, value: str) that temporarily modifies os.environ and restores the prior value upon block exit.
3. Use ExitStack to acquire a list of simulated thread locks and verify they are all released if any single acquisition fails.
4. Implement a context manager that catches ZeroDivisionError and logs a warning, while propagating all other exceptions.

---

Further reading

• PEP 343: The “with” Statement.
• Python Standard Library: contextlib documentation.
• Python Data Model: With Statement Context Managers.