Python Context Managers and the with Statement Explained

Python Context Managers

The with statement ensures resources are properly acquired and released — even if exceptions occur. File handles, database connections, locks, and timers all benefit from context managers.

Built-in Context Managers

# File handling — file ALWAYS closed even if exception occurs
with open('data.txt', 'r') as f:
    content = f.read()
# f is automatically closed here

# Multiple context managers
with open('input.txt', 'r') as src, open('output.txt', 'w') as dst:
    dst.write(src.read())

# Threading lock
import threading
lock = threading.Lock()
with lock:
    # critical section — lock released even on exception
    shared_resource.update()

# Decimal precision
from decimal import Decimal, localcontext
with localcontext() as ctx:
    ctx.prec = 50  # 50 decimal places for this block only
    result = Decimal('1') / Decimal('3')
    print(result)  # 0.33333333333333333333333333333333333333333333333333

Class-Based Context Manager

class DatabaseConnection:
    def __init__(self, host, db):
        self.host = host
        self.db   = db
        self.conn = None

    def __enter__(self):
        print(f"Connecting to {self.db}@{self.host}")
        self.conn = create_connection(self.host, self.db)
        return self.conn   # returned to the 'as' variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f"Exception occurred: {exc_val}")
            self.conn.rollback()
        else:
            self.conn.commit()
        self.conn.close()
        print("Connection closed")
        return False  # False: don't suppress exceptions; True: suppress

with DatabaseConnection('localhost', 'ezycoders') as conn:
    conn.execute('INSERT INTO users VALUES (...)')
# commit + close happen automatically

contextlib — Function-Based Context Manager

from contextlib import contextmanager
import time

@contextmanager
def timer(label=''):
    start = time.perf_counter()
    try:
        yield  # code in the with block runs here
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.4f}s")

with timer('Processing'):
    result = sum(range(10_000_000))
# Processing: 0.3421s

@contextmanager
def temp_directory():
    import tempfile, shutil, os
    tmpdir = tempfile.mkdtemp()
    try:
        yield tmpdir          # provide temp dir path
    finally:
        shutil.rmtree(tmpdir) # always cleanup

with temp_directory() as tmp:
    # do work in tmp
    with open(f'{tmp}/output.csv', 'w') as f:
        f.write('name,score\nRahul,95')
# tmp directory deleted automatically

Nested and Reusable Context Managers

from contextlib import contextmanager, suppress

# suppress — ignore specific exceptions
with suppress(FileNotFoundError):
    os.remove('might_not_exist.txt')  # no exception raised if missing

# ExitStack — dynamic context managers
from contextlib import ExitStack

files = ['a.txt', 'b.txt', 'c.txt']
with ExitStack() as stack:
    handles = [stack.enter_context(open(f)) for f in files]
    data = [h.read() for h in handles]
# all files closed

# Real-world: database transaction
@contextmanager
def transaction(conn):
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise

with transaction(db_conn) as conn:
    conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id = 1')
    conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 2')