Python Decorators and Context Managers, Measured

Series: Python lecture 4/6 (intermediate). For developers who use @something and with every day but have never written one: after this lecture you can write decorators with and without arguments, know exactly what functools.wraps fixes, and build context managers that clean up correctly even when the body fails.

If your framework routes stopped matching after you added a logging decorator, if a "safe" wrapper turned a JSON error into a confusing NoneType crash three calls later, or if a with block swallowed an exception you needed to see, this lecture is for you. All three are one-line mistakes, and each one is shown below with its real output. Everything was run on Python 3.9 (timings on an Apple Silicon Mac, M1 Pro) and the behavior was checked again on Python 3.11.

The headline numbers: lru_cache made recursive fib(30) about 32,000 times faster (175.9 ms → 5.5 µs); a do-nothing decorator roughly doubled the cost of a tiny call (89 → 180 ns); and lru_cache halved the recursion depth you can reach before RecursionError (996 → 497 frames).

Functions are objects

A decorator is only possible because a function is an ordinary object: you can assign it to another name, put it in a dict, pass it as an argument, return it from another function, and attach attributes to it.

def shout(text):
    """Return text in upper case."""
    return text.upper() + "!"

yell = shout
print(yell("hi"), yell is shout, yell.__name__)
print(type(shout), shout.__doc__)
handlers = {"loud": shout, "quiet": str.lower}
print(handlers["loud"]("ok"), handlers["quiet"]("OK"))

def make_counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc
c = make_counter()
print(c(), c(), c.__closure__[0].cell_contents)

HI! True shout
<class 'function'> Return text in upper case.
OK! ok
1 2 2

Note yell.__name__ is still shout: the name belongs to the function object, not to the variable. The last line shows a closure: inc keeps count alive in a cell after make_counter returned. Lecture 2 covered nonlocal and scope; a decorator is exactly this pattern, with a function as the captured variable.

Writing a decorator: timing

@timed above a def means func = timed(func), run once, at definition time. timed receives the original function and returns a replacement, usually a closure called wrapper that does something before and after calling the original.

import functools, time

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            ms = (time.perf_counter() - t0) * 1000
            print(f"[timed] {func.__name__} {ms:.1f} ms")
    return wrapper

@timed
def build_index(n):
    """Map each word to its positions."""
    index = {}
    for i in range(n):
        index.setdefault(f"w{i % 1000}", []).append(i)
    return index

idx = build_index(200_000)
print(len(idx), build_index.__name__, build_index.__doc__)

# what @timed means, without the syntax:
def build_index2(n): return n
build_index2 = timed(build_index2)
build_index2(1)

[timed] build_index 39.7 ms
1000 build_index Map each word to its positions.
[timed] build_index2 0.0 ms

Three habits are baked in: *args, **kwargs so any signature passes through; return func(...) so the result is not lost; and try/finally so the time is printed even when the function raises.

functools.wraps: what breaks without it

Remove the @functools.wraps(func) line and the returned object is wrapper, with wrapper's metadata:

import functools, inspect, time

def timed_bad(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@timed_bad
def load_users(limit: int = 10) -> list:
    """Load up to `limit` users."""
    return list(range(limit))

@timed
def load_items(limit: int = 10) -> list:
    """Load up to `limit` items."""
    return list(range(limit))

for f in (load_users, load_items):
    print(f.__name__, "|", f.__doc__, "|", inspect.signature(f), "|", f.__qualname__)

wrapper | None | (*args, **kwargs) | timed_bad.<locals>.wrapper
load_items | Load up to `limit` items. | (limit: int = 10) -> list | load_items

Name, docstring, signature and qualified name are all gone without wraps. help(), IDE tooltips, Sphinx, pytest's output and your logs all read these. It gets worse when something keys on the name, as many web frameworks and plugin registries do:

import functools, pickle

def logged(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

ROUTES = {}
def route(func):
    ROUTES[func.__name__] = func
    return func

@route
@logged
def home(): return "home"

@route
@logged
def about(): return "about"

print(ROUTES)
print(ROUTES["wrapper"]())
pickle.dumps(home)

{'wrapper': <function logged.<locals>.wrapper at 0x1051f09d0>}
about
Traceback (most recent call last):
  File "~/demo/nowraps2.py", line 23, in <module>
    pickle.dumps(home)
AttributeError: Can't pickle local object 'logged.<locals>.wrapper'

Two routes went in, one came out, and no error was raised: home was silently overwritten by about. The pickle failure matters as soon as you hand the function to multiprocessing (lecture 5). Add @functools.wraps(func) above def wrapper and rerun, replacing the last two lines with print(ROUTES["home"](), ROUTES["about"]()) and print(len(pickle.dumps(home)), "bytes pickled"):

{'home': <function home at 0x1011dc820>, 'about': <function about at 0x1011dc940>}
home about
32 bytes pickled

Both routes survive and pickling works, because pickle stores functions by module and qualified name, and wraps copies __module__, __name__, __qualname__, __doc__, __annotations__ and __dict__, and sets __wrapped__ to the original so inspect can see through it.

Note the order: decorators apply bottom-up, so @route received the output of @logged.

Decorators with arguments: retry

@retry(times=4) is a call that must return a decorator. That is three nested functions: the factory takes the options, the decorator takes the function, and the wrapper takes the call's arguments.

import functools, random, time

def retry(times=3, delay=0.01, exceptions=(Exception,)):
    """Decorator factory: retry(...) returns the real decorator."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    print(f"  {func.__name__}: attempt {attempt} failed: {e!r}")
                    if attempt == times:
                        raise
                    time.sleep(delay * 2 ** (attempt - 1))
        return wrapper
    return decorator

rng = random.Random(3)

@retry(times=4, exceptions=(ConnectionError,))
def fetch(url):
    if rng.random() < 0.6:
        raise ConnectionError("reset by peer")
    return f"200 OK {url}"

print(fetch("https://example.com/api"))

@retry(times=2, exceptions=(ConnectionError,))
def parse(text):
    return int(text)

try:
    parse("12a")
except ValueError as e:
    print("not retried:", e)

@retry
def oops():
    return 1
print(oops)
try:
    oops()
except TypeError as e:
    print("TypeError:", e)

  fetch: attempt 1 failed: ConnectionError('reset by peer')
  fetch: attempt 2 failed: ConnectionError('reset by peer')
  fetch: attempt 3 failed: ConnectionError('reset by peer')
200 OK https://example.com/api
not retried: invalid literal for int() with base 10: '12a'
<function retry.<locals>.decorator at 0x102e4c550>
TypeError: decorator() missing 1 required positional argument: 'func'

Points worth copying: retry only the exceptions that are worth retrying (a ValueError from bad input will fail the same way every time), back off exponentially, and re-raise with a bare raise on the last attempt so the original traceback survives. The last block is the classic slip: @retry without parentheses passes oops in as times, so oops becomes the inner decorator, and the error only appears when someone calls it. On Python 3.11 the message names it more clearly: retry.<locals>.decorator() missing 1 required positional argument: 'func'.

Caching: a hand-written memoize, then lru_cache

A cache decorator is a dict in the closure:

import functools

def memoize(func):
    cache = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache                  # functions can carry attributes
    return wrapper

@memoize
def slow_square(x):
    print(f"  computing {x}")
    return x * x

print(slow_square(4), slow_square(4), slow_square.cache)

  computing 4
16 16 {(4,): 16}

It has no size limit, ignores keyword arguments and keeps no statistics. The standard library's functools.lru_cache handles all of that in C. Here is the measurement against naive recursion, cold cache on every run:

"""Naive fib vs lru_cache fib (Python 3.9, Apple M1 Pro)."""
import functools, sys, time

def fib_plain(n):
    return n if n < 2 else fib_plain(n - 1) + fib_plain(n - 2)

@functools.lru_cache(maxsize=None)
def fib_cached(n):
    return n if n < 2 else fib_cached(n - 1) + fib_cached(n - 2)

def best(fn, arg, repeat=5, clear=None):
    out = float("inf")
    for _ in range(repeat):
        if clear: clear()                      # cold cache every run
        t0 = time.perf_counter(); fn(arg); out = min(out, time.perf_counter() - t0)
    return out

print("Python", sys.version.split()[0])
for n in (20, 25, 30):
    tp = best(fib_plain, n)
    tc = best(fib_cached, n, clear=fib_cached.cache_clear)
    print(f"fib({n}) = {fib_plain(n):>6}  plain {tp*1000:7.2f} ms  "
          f"lru_cache {tc*1e6:5.1f} us  x{tp/tc:,.0f}")

fib_cached.cache_clear()
fib_cached(30)
print(fib_cached.cache_info())
fib_cached(30)
print(fib_cached.cache_info())

Python 3.9.6
fib(20) =   6765  plain    1.43 ms  lru_cache   3.9 us  x370
fib(25) =  75025  plain   15.78 ms  lru_cache   4.9 us  x3,210
fib(30) = 832040  plain  175.90 ms  lru_cache   5.5 us  x31,740
CacheInfo(hits=28, misses=31, maxsize=None, currsize=31)
CacheInfo(hits=29, misses=31, maxsize=None, currsize=31)

A second run gave 1.44/16.16/175.96 ms and 3.7/4.8/5.5 µs. The plain version grows by about 11x for every +5 (it makes 2,692,537 calls for fib(30)); the cached one grows linearly. cache_info() is the proof: 31 misses for the 31 distinct arguments 0..30, 28 hits for the second branch of each call, and the repeat call is one more hit.

Two lru_cache traps

@functools.lru_cache(maxsize=128)
def total(items):
    return sum(items)
print(total((1, 2, 3)))
total([1, 2, 3])

6
Traceback (most recent call last):
  File "~/demo/memo.py", line 24, in <module>
    total([1, 2, 3])
TypeError: unhashable type: 'list'

Arguments become dict keys, so they must be hashable: pass tuples or frozensets.

The second trap was a surprise. Each recursive call through lru_cache uses two levels of the recursion limit instead of one. Measuring the deepest n that still works with a simple f(n) = 1 + f(n - 1), with the default limit of 1000:

recursionlimit 1000
plain max depth  996
cached max depth 497

Python 3.11 gave 997 and 498. So a memoized recursive function from a cold cache hits RecursionError: maximum recursion depth exceeded at half the depth you might expect. Warm the cache bottom-up (call f(200), f(400), …) or rewrite as a loop.

Context managers: the with protocol

with EXPR as x: calls EXPR.__enter__() and binds the result to x, runs the body, then always calls __exit__(exc_type, exc, tb), with three Nones if the body succeeded. If __exit__ returns a true value, the exception is suppressed.

import contextlib, os, tempfile, time

class Timer:
    def __enter__(self):
        self.t0 = time.perf_counter()
        print("enter")
        return self                       # bound to the name after `as`
    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self.t0
        print(f"exit: exc_type={exc_type.__name__ if exc_type else None}")
        return False                      # do not swallow the exception

with Timer() as t:
    sum(range(100_000))
print("elapsed set:", t.elapsed > 0)

try:
    with Timer():
        1 / 0
except ZeroDivisionError:
    print("ZeroDivisionError still reached the caller")

enter
exit: exc_type=None
elapsed set: True
enter
exit: exc_type=ZeroDivisionError
ZeroDivisionError still reached the caller

For most cases, contextlib.contextmanager turns a generator into a context manager: code before yield is __enter__, the yielded value is what as binds, and code after is __exit__. The try/finally is not optional:

@contextlib.contextmanager
def chdir_tmp():
    old = os.getcwd()
    path = tempfile.mkdtemp()
    os.chdir(path)
    try:
        yield path
    finally:                              # without try/finally, an error skips cleanup
        os.chdir(old)
        os.rmdir(path)

Without it, an exception in the body is re-raised at the yield, and everything after the yield is skipped:

@contextlib.contextmanager
def opened_resource(name):
    print(f"acquire {name}")
    yield name
    print(f"release {name}")          # BUG: no try/finally

try:
    with opened_resource("db-conn"):
        raise ValueError("query failed")
except ValueError as e:
    print("caught:", e)

acquire db-conn
caught: query failed

No "release" line: in real code, that is a leaked connection or lock.

contextlib.suppress is the explicit, narrow way to ignore one exception type, and ExitStack manages a number of resources that is only known at run time, closing them in reverse order:

with contextlib.suppress(FileNotFoundError):
    os.remove("does-not-exist.txt")
print("suppress: continued")

paths = []
for i in range(3):
    fd, p = tempfile.mkstemp(suffix=f"-{i}.txt"); os.close(fd); paths.append(p)
with contextlib.ExitStack() as stack:
    files = [stack.enter_context(open(p, "w")) for p in paths]
    stack.callback(print, "callback: runs last-in, first-out")
    for i, f in enumerate(files):
        f.write(f"part {i}\n")
    print("open:", [f.closed for f in files])
print("closed:", [f.closed for f in files])
for p in paths: os.remove(p)

suppress: continued
open: [False, False, False]
callback: runs last-in, first-out
closed: [True, True, True]

The callback was registered last, so it ran first, before the three files were closed. If opening the third file had failed, the first two would still have been closed.

Common mistakes: the decorator that eats results and errors

Two bugs that look harmless in review:

import functools, logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

def logged(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logging.info("call %s%r", func.__name__, args)
        func(*args, **kwargs)                       # BUG 1: result is dropped
    return wrapper

def safe(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:                      # BUG 2: swallowed, returns None
            logging.error("%s failed: %s", func.__name__, e)
    return wrapper

@logged
def price_with_tax(yen):
    return round(yen * 1.1)

total = price_with_tax(1000)
print("total:", total)

@safe
def load_config(text):
    import json
    return json.loads(text)

cfg = load_config('{"retries": 3,}')
print(cfg["retries"])

INFO call price_with_tax(1000,)
total: None
ERROR load_config failed: Expecting property name enclosed in double quotes: line 1 column 15 (char 14)
Traceback (most recent call last):
  File "~/demo/deco_bad.py", line 33, in <module>
    print(cfg["retries"])
TypeError: 'NoneType' object is not subscriptable

What actually happens:

• Forgotten return: every decorated function now returns None. No error at all; total: None flows on into the next calculation or into a JSON response.
• Swallowed exception: the real cause (a trailing comma in JSON) is logged once at the point of failure, and the crash happens later, somewhere else, as a TypeError about NoneType. In a busy log, nobody connects the two lines.

Fix: always return func(*args, **kwargs). In a decorator that catches, either re-raise (except Exception: log(...); raise) or catch only specific, expected exceptions and return a value that callers are designed to handle. The same rule holds for __exit__:

class Swallow:
    def __enter__(self): return self
    def __exit__(self, *exc):
        print("cleanup")
        return True                    # BUG: True means "exception handled"

with Swallow():
    raise KeyError("user_id")
print("the KeyError vanished")

cleanup
the KeyError vanished

Return True from __exit__ only on purpose, and only after checking exc_type.

Decision table

And one cost to remember: a pass-through decorator around a two-argument function took 178–183 ns per call against 89–90 ns for the bare function (1,000,000 calls, best of 5, two runs). That is nothing for a network call and noticeable inside a tight loop over millions of items.

Honest limits / version notes

• Timings: Python 3.9.6 on an Apple Silicon Mac (M1 Pro), time.perf_counter, best of 5, each run twice; seeds fixed (random.Random(3) for the retry demo). Python 3.11 was used only to check behavior and messages, never for timing.
• functools.cache (unbounded lru_cache) exists since 3.9 and @lru_cache without parentheses since 3.8; both confirmed on 3.9. contextlib.chdir exists since 3.11 (confirmed missing on 3.9, present on 3.11), which is why the example above writes its own.
• The recursion-depth figures (996/497 on 3.9, 997/498 on 3.11) are specific to these versions. Python 3.12 reworked how recursion through C code is limited, so the numbers may differ there; 3.12+ was not run.
• Python 3.11 tracebacks add ^^^^ markers under the failing expression; the messages were otherwise the same.

Series navigation: previous: Python lecture 3: iterators, generators and memory · next: Python lecture 5: threads, processes, asyncio and the GIL. If you come from C#, decorators fill the role that delegates and attributes share there, and with is C#'s using; see also C# exception handling for the swallowed-exception problem on the .NET side.