Context Managers — with Statement, __enter__/__exit__, contextlib

The with statement is Python's core pattern for safely acquiring and releasing resources. It is used anywhere cleanup is required: files, database connections, locks, temporary directories, and more.

---

How the with Statement Works

# Basic structure of the with statement
with expression as variable:
    body

# The code above behaves identically to:
manager = expression
variable = manager.__enter__()
try:
    body
except:
    if not manager.__exit__(*sys.exc_info()):
        raise
else:
    manager.__exit__(None, None, None)

Key Rules

• __enter__ acquires the resource and returns the value to bind to the as variable.
• __exit__ is always called when the block ends (regardless of exceptions).
• If __exit__ returns True, it suppresses the exception.

---

Implementing __enter__ / __exit__ Protocol Directly

class ManagedFile:
    """File context manager"""

    def __init__(self, path: str, mode: str = "r", encoding: str = "utf-8") -> None:
        self.path = path
        self.mode = mode
        self.encoding = encoding
        self.file = None

    def __enter__(self):
        self.file = open(self.path, self.mode, encoding=self.encoding)
        return self.file   # Bound to the as variable

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """
        exc_type:  Exception class (None if no exception)
        exc_val:   Exception instance (None if no exception)
        exc_tb:    Traceback object (None if no exception)
        Return:    True → suppress exception, False/None → propagate exception
        """
        if self.file:
            self.file.close()
        # Return False: don't suppress, propagate the exception
        return False

# Usage
with ManagedFile("data.txt") as f:
    content = f.read()

Using __exit__ Parameters

class SafeWrite:
    """Context manager that automatically deletes the file on exception"""

    def __init__(self, path: str) -> None:
        self.path = path
        self.file = None

    def __enter__(self):
        self.file = open(self.path, "w", encoding="utf-8")
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        self.file.close()
        if exc_type is not None:
            # Delete incomplete file on exception
            import os
            os.unlink(self.path)
            print(f"File deleted due to error: {self.path}")
        return False  # Propagate exception

---

contextlib.contextmanager Decorator

Using contextlib.contextmanager, you can write a context manager concisely as a generator function.

from contextlib import contextmanager

@contextmanager
def managed_file(path: str, mode: str = "r"):
    file = open(path, mode, encoding="utf-8")
    try:
        yield file          # ← Before yield: __enter__
    finally:
        file.close()        # ← After yield: __exit__

# Usage
with managed_file("data.txt") as f:
    print(f.read())

What comes before yield corresponds to __enter__, and what comes after to __exit__. The value passed to yield is bound to the as variable.

Pattern with Exception Handling

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    """Database transaction context manager"""
    try:
        yield conn
        conn.commit()           # Commit on success
    except Exception:
        conn.rollback()         # Rollback on exception
        raise                   # Re-raise exception
    finally:
        conn.close()            # Always close connection

---

contextlib.suppress — Ignoring Exceptions

Use this to silently ignore specific exceptions.

from contextlib import suppress
import os

# Without suppress
try:
    os.remove("temp.txt")
except FileNotFoundError:
    pass

# With suppress — more concise
with suppress(FileNotFoundError):
    os.remove("temp.txt")

# Suppress multiple exceptions
with suppress(FileNotFoundError, PermissionError):
    os.remove("protected.txt")

Warning: suppress completely ignores exceptions, so don't use it when you need to know about errors.

---

contextlib.ExitStack — Dynamic Context Stack

Use this when the number of files is determined dynamically or you need to conditionally activate context managers.

from contextlib import ExitStack

# Open multiple files at once (dynamic count)
def merge_files(input_paths: list[str], output_path: str) -> None:
    with ExitStack() as stack:
        # Dynamically add files to the stack
        files = [
            stack.enter_context(open(p, "r", encoding="utf-8"))
            for p in input_paths
        ]
        out = stack.enter_context(open(output_path, "w", encoding="utf-8"))

        for f in files:
            out.write(f.read())

# Conditional context
def process(data, use_lock: bool = False):
    with ExitStack() as stack:
        if use_lock:
            import threading
            lock = threading.Lock()
            stack.enter_context(lock)
        # Code below is the same regardless of use_lock
        do_work(data)

Registering Callbacks with ExitStack

from contextlib import ExitStack

with ExitStack() as stack:
    # Register callbacks to be called on exit
    stack.callback(print, "Cleanup 1 complete")
    stack.callback(print, "Cleanup 2 complete")
    print("Performing work")

# Output:
# Performing work
# Cleanup 2 complete  ← LIFO order
# Cleanup 1 complete

---

Practical Example 1 — DB Transaction

import sqlite3
from contextlib import contextmanager
from typing import Generator

@contextmanager
def db_transaction(db_path: str) -> Generator[sqlite3.Connection, None, None]:
    """
    Automatically manages SQLite transactions.
    Rolls back on exception, commits on normal exit.
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    except sqlite3.Error as e:
        conn.rollback()
        raise RuntimeError(f"DB transaction failed: {e}") from e
    finally:
        conn.close()

# Usage
def transfer_points(db: str, from_id: int, to_id: int, amount: int):
    with db_transaction(db) as conn:
        conn.execute(
            "UPDATE users SET points = points - ? WHERE id = ?",
            (amount, from_id)
        )
        conn.execute(
            "UPDATE users SET points = points + ? WHERE id = ?",
            (amount, to_id)
        )
        # Both queries roll back on exception

---

Practical Example 2 — Timer

import time
from contextlib import contextmanager
from dataclasses import dataclass, field

@dataclass
class TimerResult:
    elapsed: float = 0.0
    label: str = ""

    def __str__(self) -> str:
        return f"[{self.label}] {self.elapsed:.4f}s"

@contextmanager
def timer(label: str = ""):
    result = TimerResult(label=label)
    start = time.perf_counter()
    try:
        yield result
    finally:
        result.elapsed = time.perf_counter() - start
        print(result)

# Usage
with timer("Data processing") as t:
    # Work to process
    data = list(range(1_000_000))
    total = sum(data)

# [Data processing] 0.0312s

---

Practical Example 3 — Temporary Directory

import shutil
import tempfile
from contextlib import contextmanager
from pathlib import Path

@contextmanager
def temp_directory(prefix: str = "tmp_"):
    """Creates a temporary directory and automatically deletes it on exit."""
    tmp_dir = Path(tempfile.mkdtemp(prefix=prefix))
    try:
        yield tmp_dir
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)

# Usage
with temp_directory("build_") as tmp:
    (tmp / "output.txt").write_text("Temporary file")
    (tmp / "subdir").mkdir()
    # Entire tmp directory is deleted when the with block exits

# Standard library: tempfile.TemporaryDirectory()
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
    path = Path(tmpdir) / "file.txt"
    path.write_text("Content")

---

Async Context Manager (async with)

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_timer(label: str = ""):
    start = asyncio.get_event_loop().time()
    try:
        yield
    finally:
        elapsed = asyncio.get_event_loop().time() - start
        print(f"[{label}] {elapsed:.4f}s")

async def main():
    async with async_timer("Async operation"):
        await asyncio.sleep(0.1)

asyncio.run(main())

---

Expert Tips

Tip 1 — Pattern of returning self from __enter__

class Connection:
    def __enter__(self):
        self.connect()
        return self   # Return self to enable: with ... as conn: conn.method()

    def __exit__(self, *args):
        self.close()
        return False

Tip 2 — Using contextlib.contextmanager with async

asynccontextmanager is the async version of contextmanager. It is essential for code using async for and async with.

Tip 3 — Use a single with instead of nested with (Python 3.10+)

# Old style
with open("a.txt") as f1:
    with open("b.txt") as f2:
        pass

# Python 3.10+ parenthesis syntax
with (
    open("a.txt") as f1,
    open("b.txt") as f2,
):
    pass

Tip 4 — nullcontext — Conditional context manager

from contextlib import nullcontext

def process(data, lock=None):
    # Use a no-op context if lock is None
    with lock if lock is not None else nullcontext():
        do_work(data)