Context Managers

Context Managers

Context managers handle setup and teardown automatically — most commonly used for file I/O, database connections, and locks.

The with Statement

# Without context manager
f = open("file.txt", "r")
try:
    data = f.read()
finally:
    f.close()

# With context manager — cleaner and safer
with open("file.txt", "r") as f:
    data = f.read()
# f is automatically closed here, even if an exception occurs

Multiple Context Managers

with open("input.txt") as fin, open("output.txt", "w") as fout:
    fout.write(fin.read())

How It Works: __enter__ and __exit__

class ManagedResource:
    def __enter__(self):
        print("Acquiring resource")
        return self   # value bound to 'as' variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Releasing resource")
        # Return True to suppress exceptions, False to propagate
        return False

with ManagedResource() as r:
    print("Using resource")
# Output:
# Acquiring resource
# Using resource
# Releasing resource

Exception Handling in __exit__

class SafeDiv:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ZeroDivisionError:
            print("Caught division by zero!")
            return True   # suppress the exception
        return False

with SafeDiv():
    result = 10 / 0   # exception suppressed
print("Continues here")

contextlib.contextmanager

The easiest way to create a context manager using a generator:

from contextlib import contextmanager

@contextmanager
def timer():
    import time
    start = time.time()
    try:
        yield   # code inside 'with' block runs here
    finally:
        elapsed = time.time() - start
        print(f"Elapsed: {elapsed:.3f}s")

with timer():
    sum(range(1_000_000))

contextlib.suppress

Suppress specific exceptions:

from contextlib import suppress

with suppress(FileNotFoundError):
    open("nonexistent.txt")
# No exception raised — silently ignored

contextlib.redirect_stdout

from contextlib import redirect_stdout
import io

f = io.StringIO()
with redirect_stdout(f):
    print("This goes to the buffer")

output = f.getvalue()
print(output)  # "This goes to the buffer\n"

Database Connection Pattern

from contextlib import contextmanager
import sqlite3

@contextmanager
def get_db(path):
    conn = sqlite3.connect(path)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

with get_db("app.db") as db:
    db.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))

Threading Lock

import threading

lock = threading.Lock()

with lock:
    # critical section — lock acquired automatically
    shared_resource += 1
# lock released automatically