Python Context Managers

The with Statement

The with statement ensures that setup and teardown code always runs, even if an exception occurs.

# Without with — error-prone
f = open("data.txt")
try:
    data = f.read()
finally:
    f.close()  # must remember to close

# With with — clean and safe
with open("data.txt") as f:
    data = f.read()
# f.close() is called automatically, even on exception

Every object used in a with statement implements the context manager protocol:

• __enter__(self) — called on entry, return value bound to as target
• __exit__(self, exc_type, exc_val, exc_tb) — called on exit; return truthy to suppress the exception

Multiple Context Managers

# Equivalent to nested with statements
with open("input.txt") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

# Python 3.10+ — parenthesized for long lines
with (
    open("input.txt") as src,
    open("output.txt", "w") as dst,
    threading.Lock() as lock,
):
    dst.write(src.read())

Building a Class-Based Context Manager

import time
import logging

logger = logging.getLogger(__name__)

class Timer:
    """Measure elapsed time for a block of code."""

    def __init__(self, label: str = ""):
        self.label = label
        self.elapsed: float = 0.0

    def __enter__(self):
        self._start = time.perf_counter()
        return self  # bound to 'as' target

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self._start
        logger.debug("%s: %.4fs", self.label, self.elapsed)
        return False  # do not suppress exceptions

with Timer("database query") as t:
    results = db.execute("SELECT * FROM users")

print(f"Query took {t.elapsed:.4f}s")

Suppressing Exceptions

Return True from __exit__ to suppress an exception:

class SuppressError:
    """Suppress a specific exception type."""

    def __init__(self, *exc_types):
        self.exc_types = exc_types

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        return exc_type is not None and issubclass(exc_type, self.exc_types)

with SuppressError(FileNotFoundError):
    os.remove("maybe_exists.tmp")
# No error even if file doesn't exist

contextlib.contextmanager

For simple cases, write a generator function instead of a full class:

from contextlib import contextmanager
import os

@contextmanager
def working_directory(path: str):
    """Temporarily change the working directory."""
    original = os.getcwd()
    try:
        os.chdir(path)
        yield  # control passes to the with block
    finally:
        os.chdir(original)  # always restore

with working_directory("/tmp"):
    print(os.getcwd())  # /tmp

print(os.getcwd())  # back to original

With a return value:

from contextlib import contextmanager
import sqlite3

@contextmanager
def db_transaction(connection):
    """Run a block inside a transaction; commit on success, rollback on error."""
    cursor = connection.cursor()
    try:
        yield cursor
        connection.commit()
    except Exception:
        connection.rollback()
        raise

with db_transaction(conn) as cur:
    cur.execute("INSERT INTO orders VALUES (?, ?)", (1, "book"))
    cur.execute("UPDATE inventory SET qty = qty - 1 WHERE item = 'book'")
# Committed only if both statements succeeded

contextlib Utilities

suppress

from contextlib import suppress
import os

# Instead of try/except just for cleanup
with suppress(FileNotFoundError):
    os.remove("temp.txt")

with suppress(KeyError, IndexError):
    value = data["key"][0]

redirect_stdout / redirect_stderr

from contextlib import redirect_stdout
import io

output = io.StringIO()
with redirect_stdout(output):
    print("This goes to the StringIO buffer")
    help(len)   # normally prints to stdout

captured = output.getvalue()

ExitStack

ExitStack manages a dynamic number of context managers:

from contextlib import ExitStack

files_to_process = ["a.txt", "b.txt", "c.txt"]

with ExitStack() as stack:
    file_handles = [
        stack.enter_context(open(f))
        for f in files_to_process
    ]
    # All files are open here
    for fh in file_handles:
        process(fh.read())
# All files closed here, even if some processing failed

# Also useful for conditional context managers
with ExitStack() as stack:
    if need_lock:
        stack.enter_context(threading.Lock())
    if debug_mode:
        stack.enter_context(Timer("operation"))
    do_work()

asynccontextmanager

from contextlib import asynccontextmanager
import aiohttp

@asynccontextmanager
async def http_session():
    async with aiohttp.ClientSession() as session:
        yield session

async def main():
    async with http_session() as session:
        response = await session.get("https://api.example.com/data")
        data = await response.json()

Reentrant Context Managers

The standard Lock is not reentrant — a thread trying to acquire a lock it already holds will deadlock. Use RLock:

import threading

lock = threading.Lock()
rlock = threading.RLock()

# Deadlock with Lock
with lock:
    with lock:   # blocks forever — same thread, same lock
        pass

# Safe with RLock
with rlock:
    with rlock:  # works — counts acquisitions
        pass

contextlib.contextmanager generators are reentrant-safe if written carefully.

Real-World Patterns

Managed Database Connection Pool

from contextlib import contextmanager
from queue import Queue

class ConnectionPool:
    def __init__(self, size: int, factory):
        self._pool = Queue(maxsize=size)
        for _ in range(size):
            self._pool.put(factory())

    @contextmanager
    def acquire(self):
        conn = self._pool.get()
        try:
            yield conn
        finally:
            self._pool.put(conn)

pool = ConnectionPool(5, lambda: create_db_connection())

with pool.acquire() as conn:
    conn.execute("SELECT 1")

Temporary Configuration Override

from contextlib import contextmanager
from typing import Any

class Config:
    debug = False
    log_level = "INFO"

@contextmanager
def override_config(**kwargs: Any):
    original = {k: getattr(Config, k) for k in kwargs}
    try:
        for k, v in kwargs.items():
            setattr(Config, k, v)
        yield
    finally:
        for k, v in original.items():
            setattr(Config, k, v)

with override_config(debug=True, log_level="DEBUG"):
    run_tests()   # runs with debug enabled
# Config restored to original values