Iterators, generators, closures and decorators · about 45 min

Context managers

Why this matters

Many resources must be released after use: files must be closed, locks released, database transactions committed or rolled back, temporary changes undone. The hard part is doing it on every way out, including when an exception flies through. try/finally works but clutters code, and it is easy to forget. The with statement packages the setup and the guaranteed cleanup into a reusable object, a context manager. You have used with open(...) since the files lesson; now you will see the protocol behind it and write your own, both as classes and with a generator shortcut.

What with does

with open("notes.txt", "w") as f:
    f.write("hello\n")
# f is closed here, whatever happened inside

The with statement is roughly this code:

mgr = open("notes.txt", "w")
f = mgr.__enter__()               # setup; the result is bound by "as"
try:
    f.write("hello\n")            # the body
except BaseException as e:
    if not mgr.__exit__(type(e), e, e.__traceback__):
        raise                     # not suppressed: continue propagating
else:
    mgr.__exit__(None, None, None)

So the protocol is two methods:

__exit__ also runs when you leave the block with return, break or continue.

A class-based context manager

class Tag:
    def __init__(self, name):
        self.name = name
    def __enter__(self):
        print(f"<{self.name}>")
        return self
    def __exit__(self, exc_type, exc, tb):
        print(f"</{self.name}>")
        return False

with Tag("html"):
    with Tag("body"):
        print("hi")
<html>
<body>
hi
</body>
</html>

Nested with blocks unwind in reverse order, like a stack. Several managers can go on one line: with Tag("a"), Tag("b"):, or, in parentheses, across lines.

Now with an exception:

try:
    with Tag("p"):
        print(1 / 0)
except ZeroDivisionError as e:
    print("caught:", e)
<p>
</p>
caught: division by zero

The closing tag still printed: __exit__ ran, returned False, and the exception continued to the outer try.

Suppressing exceptions

Returning True from __exit__ swallows the exception, and execution continues after the with block. This is powerful and dangerous; do it only for specific exception types. The standard library has a ready-made one:

import contextlib, os

with contextlib.suppress(FileNotFoundError):
    os.remove("maybe-missing.tmp")
print("carried on")
carried on

That replaces a four-line try/except FileNotFoundError: pass.

The generator shortcut: contextlib.contextmanager

Writing a class for every context manager is verbose. contextlib.contextmanager turns a generator function with a single yield into one:

from contextlib import contextmanager

@contextmanager
def tag(name):
    print(f"<{name}>")         # __enter__ part
    try:
        yield name             # the value bound by "as"; the body runs here
    finally:
        print(f"</{name}>")    # __exit__ part

with tag("b") as t:
    print("bold", t)
<b>
bold b
</b>

How it works: the decorator wraps your generator in an object whose __enter__ calls next() to run up to the yield, and whose __exit__ resumes the generator. If the body raised, __exit__ throws the exception into the generator at the yield, which is why you need try/finally (or try/except) around the yield to guarantee cleanup. Without it, an exception in the body would skip your cleanup code. To suppress an exception, catch it around the yield and do not re-raise.

A practical example, timing a block (output values vary, so we just show the shape):

import time
from contextlib import contextmanager

@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.3f}s")

with timer("sum"):
    total = sum(range(10_000_000))
sum: 0.142s

Where you will meet context managers

import io, contextlib
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
    print("captured")
print(repr(buf.getvalue()))
'captured\n'

ExitStack: a dynamic number of managers

When you do not know until run time how many resources you need, for example opening every file in a list, contextlib.ExitStack collects managers and exits them all, in reverse order, when its own block ends:

from contextlib import ExitStack

names = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
    files = [stack.enter_context(open(n, "w")) for n in names]
    for f in files:
        f.write("data\n")
# all three files are closed here, even if one open() failed halfway

stack.callback(func, *args) also registers arbitrary cleanup functions.

Common mistakes

Recap

Going further: read the contextlib documentation and PEP 343, "The with Statement", which shows the exact translation of with into try/finally.

Exercises

Exercise 1 · quiz

Does __exit__ run on errors?

What does this print before the traceback appears? Type the printed words in order, separated by single spaces.

class Box:
    def __enter__(self):
        print("open")
        return self
    def __exit__(self, exc_type, exc, tb):
        print("close")
        return False

with Box():
    print("work")
    raise ValueError("oops")
    print("more")
Answer

__enter__ runs, then the body. When the body raises, Python calls __exit__ with the exception details before the exception continues outward. Because __exit__ returns False, the exception is not suppressed and the traceback follows. more never runs. Cleanup in __exit__ happens on every way out of the block: normal completion, exception, return, break.

Exercise 2 · write code

An indentation context manager

In solution.py, write a class Indenter used like this:

ind = Indenter()
with ind:
    ind.print("level 1")
    with ind:
        ind.print("level 2")
    ind.print("back to 1")
ind.print("level 0")

which prints

    level 1
        level 2
    back to 1
level 0

Each with ind: increases the indentation by 4 spaces in __enter__ and decreases it in __exit__. ind.print(text) prints the text with the current indentation. __enter__ returns the Indenter itself (so with Indenter() as ind: works too). The level must be restored even if the body raises, and exceptions must not be swallowed.

File: solution.py

Starting point

class Indenter:
    def __init__(self):
        self.level = 0

    def print(self, text):
        print(" " * 4 * self.level + text)
Hint 1

__enter__(self) increments self.level and returns self.

Hint 2

__exit__(self, exc_type, exc, tb) decrements it and returns False (or None) so exceptions propagate.

Reference solution
class Indenter:
    def __init__(self):
        self.level = 0

    def __enter__(self):
        self.level += 1
        return self

    def __exit__(self, exc_type, exc, tb):
        self.level -= 1
        return False

    def print(self, text):
        print(" " * 4 * self.level + text)

Exercise 3 · write code

A generator-based context manager

In solution.py, use contextlib.contextmanager to write temp_setting(settings, key, value). settings is a dict. On entry it sets settings[key] = value and yields the dict. On exit it restores the old value, or deletes the key if it did not exist before, even if the body raised (and without swallowing the exception).

cfg = {"debug": False}
with temp_setting(cfg, "debug", True) as c:
    c["debug"]          # True
cfg                     # {'debug': False}
with temp_setting(cfg, "verbose", 2):
    cfg["verbose"]      # 2
cfg                     # {'debug': False}

File: solution.py

Starting point

from contextlib import contextmanager


@contextmanager
def temp_setting(settings, key, value):
    settings[key] = value
    yield settings
Hint 1

Remember the old value before changing it. Because the old value could legitimately be None, use a unique sentinel object such as _MISSING = object() for "not present".

Hint 2

Wrap the yield in try: ... finally: so the restore runs even when the body raises.

Reference solution
from contextlib import contextmanager

_MISSING = object()


@contextmanager
def temp_setting(settings, key, value):
    old = settings.get(key, _MISSING)
    settings[key] = value
    try:
        yield settings
    finally:
        if old is _MISSING:
            del settings[key]
        else:
            settings[key] = old

Exercise 4 · write code

Suppressing exceptions on purpose

In solution.py, write a class-based context manager Ignore(*exc_types) that suppresses exceptions of the given types raised inside the block (like contextlib.suppress) and records them: after the block, the attribute caught holds the suppressed exception object, or None if nothing was suppressed. Exceptions of other types must propagate. Do not use contextlib.suppress.

with Ignore(KeyError, IndexError) as ig:
    {}["missing"]
type(ig.caught)       # <class 'KeyError'>

File: solution.py

Starting point

class Ignore:
    def __init__(self, *exc_types):
        self.exc_types = exc_types
        self.caught = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return True
Hint 1

__exit__ receives exc_type as None when the block finished normally.

Hint 2

Returning a true value from __exit__ suppresses the exception; returning False lets it continue.

Hint 3

issubclass(exc_type, self.exc_types) accepts a tuple of classes, and also matches subclasses.

Reference solution
class Ignore:
    def __init__(self, *exc_types):
        self.exc_types = exc_types
        self.caught = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        if exc_type is not None and issubclass(exc_type, self.exc_types):
            self.caught = exc
            return True
        return False