Python · Syntax · Advanced

Context managers

10 tasks

Objects that manage setup and teardown with the `with` statement. Covers `__enter__`/`__exit__`, `contextlib.contextmanager`, and resource management patterns.

Quick topic start and explanations before exercises (exercises below):

__enter__, __exit__, and exception handling in depth

#
**Why context managers exist** Resource management follows a strict pattern: *acquire → use → release*. The release step must happen even if an exception occurs during 'use'. Without a guaranteed cleanup, bugs leave files open, database connections leaking, and locks never released. The naive approach with `try/finally` works but forces every caller to repeat the cleanup logic: ```python # Repetitive — every caller must write this boilerplate f = open('data.txt') try: content = f.read() finally: f.close() # must close even if f.read() raised ``` A context manager encapsulates both the setup and teardown in one reusable object, and the `with` statement calls them automatically: ```python with open('data.txt') as f: content = f.read() # file is closed here, even if f.read() raised ``` **The protocol: `__enter__` and `__exit__`** Any object with these two methods can be used in a `with` statement: ```python class Timer: def __enter__(self): import time self._start = time.perf_counter() return self # this becomes the 'as' target def __exit__(self, exc_type, exc_val, exc_tb): import time elapsed = time.perf_counter() - self._start print(f'elapsed: {elapsed:.4f}s') return False # False = do not suppress exceptions with Timer() as t: sum(range(1_000_000)) # elapsed: 0.0312s (printed by __exit__) ``` **What the three `__exit__` arguments mean** `__exit__(self, exc_type, exc_val, exc_tb)` is called when the `with` block ends — whether normally or due to an exception. - If the block ended *without* an exception: all three are `None` - If an exception *was* raised: `exc_type` is the exception class, `exc_val` is the instance, `exc_tb` is the traceback The return value controls whether the exception propagates: ```python class SuppressKeyError: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is KeyError: print(f'Suppressed missing key: {exc_val}') return True # True = swallow the exception return False # False = let it propagate d = {'a': 1} with SuppressKeyError(): print(d['missing']) # Suppressed missing key: 'missing' print('continues here') # execution resumes normally ``` Returning `True` from `__exit__` is how you suppress exceptions. Returning `False` (or `None`, which is falsy) lets them propagate. Only suppress exceptions deliberately — accidentally returning `True` hides bugs.

@contextmanager, suppress, ExitStack, and nesting

#
**`@contextmanager` — write a context manager as a generator** Writing a full class for simple context managers is verbose. `contextlib.contextmanager` lets you write one as a generator function with a single `yield`: ```python from contextlib import contextmanager @contextmanager def managed_connection(host): conn = connect(host) # __enter__: setup try: yield conn # 'as' target; code inside 'with' runs here finally: conn.close() # __exit__: teardown — always runs with managed_connection('localhost') as conn: conn.query('SELECT 1') ``` The `try/finally` around `yield` is essential. Without it, an exception inside the `with` block would skip `conn.close()` — defeating the whole purpose: ```python @contextmanager def bad_manager(): # BUG: missing try/finally resource = acquire() yield resource release(resource) # never reached if body raises! @contextmanager def good_manager(): # correct resource = acquire() try: yield resource finally: release(resource) # always runs ``` **`contextlib.suppress(*exc_types)` — inline exception suppression** Equivalent to `try/except` that catches and ignores specific exceptions: ```python from contextlib import suppress import os # Without suppress: try: os.remove('maybe_exists.txt') except FileNotFoundError: pass # With suppress: with suppress(FileNotFoundError): os.remove('maybe_exists.txt') ``` **`contextlib.ExitStack` — dynamic context manager composition** `ExitStack` manages a variable number of context managers that you don't know at write time. It also works as an escape hatch when something goes wrong during setup: ```python from contextlib import ExitStack # Open a variable-length list of files safely: filenames = ['a.txt', 'b.txt', 'c.txt'] with ExitStack() as stack: files = [stack.enter_context(open(f)) for f in filenames] # All files are open here for f in files: print(f.read()) # All files are closed here — even if one read() raised # ExitStack as a cleanup-on-failure pattern: def setup_resources(): stack = ExitStack() try: conn = stack.enter_context(get_connection()) lock = stack.enter_context(get_lock()) return conn, lock, stack # caller owns the stack except Exception: stack.close() # clean up what was acquired so far raise ``` **Nesting context managers** You can open multiple context managers in one `with` statement — they're entered left to right and exited right to left (like a stack): ```python # Both styles are identical: with open('in.txt') as src, open('out.txt', 'w') as dst: dst.write(src.read()) # Equivalent nested form: with open('in.txt') as src: with open('out.txt', 'w') as dst: dst.write(src.read()) ```

class vs @contextmanager, real patterns, nullcontext, common mistakes

#
**class vs `@contextmanager` — when to choose which** Both produce a context manager. The choice comes down to complexity and reuse: | Situation | Use | |---|---| | Simple setup + teardown, one-off | `@contextmanager` | | Need to customise `__enter__` return value in complex ways | Class | | Want to subclass or add methods | Class | | Sharing the CM across multiple threads or re-entering it | Class (be careful with state) | | Testing — you want to mock or subclass it | Class | **Real-world patterns** ```python from contextlib import contextmanager # 1. Temporary directory (stdlib already has this, but illustrative) @contextmanager def temp_dir(): import tempfile, shutil d = tempfile.mkdtemp() try: yield d finally: shutil.rmtree(d) # 2. Temporary attribute override @contextmanager def set_attr(obj, name, value): old = getattr(obj, name, None) setattr(obj, name, value) try: yield finally: if old is None: delattr(obj, name) else: setattr(obj, name, old) # 3. Indented text writer @contextmanager def indent(writer, spaces=4): writer.depth = getattr(writer, 'depth', 0) + spaces try: yield writer finally: writer.depth -= spaces ``` **`contextlib.nullcontext` — a no-op context manager** When you sometimes want a context manager and sometimes don't, `nullcontext` acts as a transparent placeholder: ```python from contextlib import nullcontext def process(data, lock=None): cm = lock if lock is not None else nullcontext() with cm: return expensive_computation(data) process(data, lock=threading.Lock()) # thread-safe process(data) # no locking — same code path ``` **Common mistakes** Missing `try/finally` in `@contextmanager` — the single most common bug. The teardown code after `yield` is skipped if the `with` body raises. Returning a value from `__exit__` by accident — any truthy value suppresses the exception. Forgetting `return False` at the bottom of `__exit__` is safe (Python treats `None` as falsy), but `return exc_tb` or `return 1` will silently swallow exceptions. Using a context manager after `with` exits — the resource is closed/released; the `as` variable still points to the object but it's in a closed state. Always complete all resource usage *inside* the `with` block.

Exercises:

01

Timer context manager

#

Write a class-based context manager `Timer` that measures elapsed time. `__enter__` should record the start time and return `self`. `__exit__` should compute and store the elapsed time in `self.elapsed`. After the `with` block, `timer.elapsed` should hold the elapsed seconds.

import time

class Timer:
    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with Timer() as t:
    total = sum(range(1_000_000))

print(f'Elapsed: {t.elapsed:.4f}s')
Solution 
import time

class Timer:
    def __enter__(self):
        self._start = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.time() - self._start
        return False  # do not suppress exceptions


with Timer() as t:
    total = sum(range(1_000_000))

print(f'Elapsed: {t.elapsed:.4f}s')
02

Managed file writer

#

Write a context manager `ManagedFile(path, mode)` using a class that opens a file in `__enter__` and closes it in `__exit__`. If an exception occurs inside the `with` block, the file should still be closed. Return the file object from `__enter__`.

class ManagedFile:
    def __init__(self, path, mode='r'):
        pass

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with ManagedFile('/tmp/test.txt', 'w') as f:
    f.write('hello context manager\n')

with ManagedFile('/tmp/test.txt', 'r') as f:
    print(f.read())  # hello context manager
Solution 
class ManagedFile:
    def __init__(self, path, mode='r'):
        self.path = path
        self.mode = mode
        self._file = None

    def __enter__(self):
        self._file = open(self.path, self.mode)
        return self._file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._file:
            self._file.close()
        return False


with ManagedFile('/tmp/test.txt', 'w') as f:
    f.write('hello context manager\n')

with ManagedFile('/tmp/test.txt', 'r') as f:
    print(f.read())
03

@contextmanager decorator

#

Rewrite the `Timer` context manager from exercise 1 using `@contextlib.contextmanager` and a generator function instead of a class. The function should `yield` a dict `{'elapsed': None}` and update it with the elapsed time after the `yield`.

import time
from contextlib import contextmanager

@contextmanager
def timer():
    pass


with timer() as t:
    total = sum(range(1_000_000))

print(f"Elapsed: {t['elapsed']:.4f}s")
Solution 
import time
from contextlib import contextmanager

@contextmanager
def timer():
    info = {'elapsed': None}
    start = time.time()
    try:
        yield info           # 'as' target receives this
    finally:
        info['elapsed'] = time.time() - start


with timer() as t:
    total = sum(range(1_000_000))

print(f"Elapsed: {t['elapsed']:.4f}s")
04

suppress_and_log context manager

#

Write a context manager `suppress_and_log(*exception_types)` that suppresses any of the given exception types and prints `'Suppressed: <exception message>'`. Other exceptions should propagate normally.

from contextlib import contextmanager

@contextmanager
def suppress_and_log(*exception_types):
    pass


with suppress_and_log(ValueError, ZeroDivisionError):
    result = 1 / 0
    print('this line is not reached')

print('execution continues after the with block')
Solution 
from contextlib import contextmanager

@contextmanager
def suppress_and_log(*exception_types):
    try:
        yield
    except exception_types as e:
        print(f'Suppressed: {e}')


with suppress_and_log(ValueError, ZeroDivisionError):
    result = 1 / 0
    print('this line is not reached')

print('execution continues after the with block')
# Suppressed: division by zero
# execution continues after the with block
05

Temporary file context manager

#

Write a context manager `temp_file(suffix='.txt')` using `@contextmanager` that creates a temporary file, yields its path, and deletes the file when the `with` block exits (even if an exception occurs). Use `tempfile.mktemp()` to generate the path.

import os
import tempfile
from contextlib import contextmanager

@contextmanager
def temp_file(suffix='.txt'):
    pass


with temp_file() as path:
    with open(path, 'w') as f:
        f.write('temporary data')
    print(os.path.exists(path))  # True

print(os.path.exists(path))  # False — file deleted
Solution 
import os
import tempfile
from contextlib import contextmanager

@contextmanager
def temp_file(suffix='.txt'):
    path = tempfile.mktemp(suffix=suffix)
    try:
        yield path
    finally:
        if os.path.exists(path):
            os.remove(path)


with temp_file() as path:
    with open(path, 'w') as f:
        f.write('temporary data')
    print(os.path.exists(path))  # True

print(os.path.exists(path))  # False
06

Indented output context manager

#

Write a context manager `indented(level=1, indent=' ')` that makes all `print()` calls inside the `with` block print with extra indentation. Patch the built-in `print` function temporarily using `builtins.print`. Restore the original `print` after the block exits.

import builtins
from contextlib import contextmanager

@contextmanager
def indented(level=1, indent='  '):
    pass


print('top level')
with indented(2):
    print('indented by 4 spaces')
    print('also indented')
print('back to top level')
Solution 
import builtins
from contextlib import contextmanager

@contextmanager
def indented(level=1, indent='  '):
    prefix = indent * level
    original_print = builtins.print
    def indented_print(*args, **kwargs):
        original_print(prefix, end='')
        original_print(*args, **kwargs)
    builtins.print = indented_print
    try:
        yield
    finally:
        builtins.print = original_print


print('top level')
with indented(2):
    print('indented by 4 spaces')
    print('also indented')
print('back to top level')
07

Transaction context manager

#

Write a class `FakeDB` with a list `log` of committed operations. Implement a context manager `transaction(db)` using `@contextmanager`: it should collect operations in a temporary list, and on clean exit commit them all to `db.log`. On exception, roll back (discard the temp list) and re-raise the exception.

from contextlib import contextmanager

class FakeDB:
    def __init__(self):
        self.log = []

@contextmanager
def transaction(db):
    pass


db = FakeDB()

with transaction(db) as tx:
    tx.append('INSERT user')
    tx.append('UPDATE balance')

print(db.log)  # ['INSERT user', 'UPDATE balance']

try:
    with transaction(db) as tx:
        tx.append('DELETE everything')
        raise RuntimeError('oops')
except RuntimeError:
    pass

print(db.log)  # still ['INSERT user', 'UPDATE balance'] — rolled back
Solution 
from contextlib import contextmanager

class FakeDB:
    def __init__(self):
        self.log = []

@contextmanager
def transaction(db):
    pending = []
    try:
        yield pending
        db.log.extend(pending)  # commit on success
    except Exception:
        pass  # rollback — discard pending
        raise


db = FakeDB()

with transaction(db) as tx:
    tx.append('INSERT user')
    tx.append('UPDATE balance')

print(db.log)  # ['INSERT user', 'UPDATE balance']

try:
    with transaction(db) as tx:
        tx.append('DELETE everything')
        raise RuntimeError('oops')
except RuntimeError:
    pass

print(db.log)  # ['INSERT user', 'UPDATE balance']
08

ExitStack for dynamic context managers

#

Use `contextlib.ExitStack` to open a dynamic number of files at once and read their first line. Given a list of file paths, open all of them inside a single `with ExitStack()` block and collect the first line from each file into a list.

import contextlib

def read_first_lines(paths):
    pass


import tempfile, os

# Create test files
paths = []
for i in range(3):
    p = tempfile.mktemp()
    with open(p, 'w') as f:
        f.write(f'line from file {i}\nmore lines')
    paths.append(p)

print(read_first_lines(paths))
# ['line from file 0', 'line from file 1', 'line from file 2']

for p in paths:
    os.remove(p)
Solution 
import contextlib

def read_first_lines(paths):
    with contextlib.ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        return [f.readline().strip() for f in files]


import tempfile, os

paths = []
for i in range(3):
    p = tempfile.mktemp()
    with open(p, 'w') as f:
        f.write(f'line from file {i}\nmore lines')
    paths.append(p)

print(read_first_lines(paths))

for p in paths:
    os.remove(p)
09

Exception-suppressing context manager

#

Write a class `Attempt` that suppresses `ValueError` exceptions raised inside the `with` block and stores the exception in `self.error`. If no exception occurred, `self.error` should be `None`. Other exception types should propagate normally.

class Attempt:
    def __init__(self):
        self.error = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with Attempt() as a:
    int('not a number')

print(a.error)  # invalid literal for int() with base 10: 'not a number'

with Attempt() as a:
    x = int('42')

print(a.error)  # None
Solution 
class Attempt:
    def __init__(self):
        self.error = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            self.error = exc_val
            return True  # suppress the exception
        return False     # propagate other exceptions


with Attempt() as a:
    int('not a number')

print(a.error)

with Attempt() as a:
    x = int('42')

print(a.error)  # None
10

Thread-safe counter with Lock

#

Write a `SafeCounter` class with an internal `threading.Lock`. Implement an `increment()` method that uses `with self.lock:` to safely increment `self.count`. Then run 5 threads, each calling `increment()` 1000 times, and verify the final count is exactly 5000.

import threading

class SafeCounter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        pass


counter = SafeCounter()
threads = [threading.Thread(target=lambda: [counter.increment() for _ in range(1000)])
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(counter.count)  # 5000
Solution 
import threading

class SafeCounter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.count += 1


counter = SafeCounter()
threads = [threading.Thread(target=lambda: [counter.increment() for _ in range(1000)])
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(counter.count)  # 5000