Build a Utility Toolkit You Will Reuse
Stop rewriting the same three helpers. Build a small helpers module with a timing decorator, a retry that survives flaky calls, a cache that skips repeated work, and a context manager that always cleans up, then reuse it in every project after this one.
About this project
Look back at what you have built. The API tracker retried a failing request. The log tool timed how long a pass took. Half of them opened something that had to be closed whether or not the code in between blew up. Each time, that logic was written again, slightly differently, tangled into the thing it was wrapping.
Why it is worth building: this project produces an actual artifact — a helpers.py you will paste into projects for years. But the real content is the shape: a decorator adds behavior to a function without touching it, which is what lets timing, retrying and caching stop being copy-paste. It is the most reusable thing in the Advanced track.
Everything here runs on this page. Nothing to install.
Build it step by step
We will build each helper from the complaint it answers, then collect them into one module at the end. Every step runs here, including the ones that fail on purpose.
Step 1: The wrapper shape, once
Every decorator in this project has the same skeleton, so it is worth meeting it on its own before anything useful happens inside.
The part people leave out is @wraps(func). Without it the wrapped function silently loses its own name and docstring, which breaks debugging, help(), and anything that inspects functions.
from functools import wraps
def announce(func):
"""The minimal decorator: do something, then call through."""
@wraps(func)
def wrapper(*args, **kwargs):
print(f"calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
@announce
def add(a, b):
"""Add two numbers."""
return a + b
print("result:", add(2, 3))
print("name kept: ", add.__name__)
print("docs kept: ", add.__doc__)
calling add result: 5 name kept: add docs kept: Add two numbers.
Step 2: A retry that survives a flaky call
The first genuinely useful one. Network calls fail intermittently, and the fix is nearly always “wait a moment and try again”, which does not belong inside the function doing the work.
This decorator takes an argument, so it needs three layers: a function that takes the settings, which returns the decorator, which returns the wrapper. Write it once, understand it once.
from functools import wraps
def retry(attempts=3, exceptions=(Exception,), sleep=None):
"""Retry a call up to `attempts` times. `sleep` is injected so tests are fast."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as err:
if attempt == attempts:
raise
print(f" {func.__name__} failed ({err}); retry {attempt}/{attempts - 1}")
if sleep:
sleep(attempt)
return wrapper
return decorator
calls = []
@retry(attempts=3)
def flaky():
"""Fails the first two times, then works."""
calls.append(1)
if len(calls) < 3:
raise ConnectionError("connection reset")
return "got the data"
print("result:", flaky())
print("attempts made:", len(calls))
flaky failed (connection reset); retry 1/2 flaky failed (connection reset); retry 2/2 result: got the data attempts made: 3
Step 3: A cache, for free
Sometimes the fix is not doing the work at all. functools.lru_cache remembers what a function returned for a given set of arguments and hands back the stored answer next time.
The rule for using it safely: only on pure functions: same input, same output, no side effects. Cache something that reads a file or the clock and you will be served a stale answer forever.
from functools import lru_cache
lookups = []
@lru_cache(maxsize=128)
def price_of(sku):
"""Pretend this is a slow database or API lookup."""
lookups.append(sku)
return {"A-1": 9.99, "B-2": 24.50}.get(sku, 0.0)
for sku in ["A-1", "B-2", "A-1", "A-1", "B-2"]:
print(f"{sku}: {price_of(sku)}")
print("real lookups:", len(lookups), "for", 5, "calls")
print(price_of.cache_info())
A-1: 9.99 B-2: 24.5 A-1: 9.99 A-1: 9.99 B-2: 24.5 real lookups: 2 for 5 calls CacheInfo(hits=3, misses=2, maxsize=128, currsize=2)
Step 4: Cleanup that survives a crash
The last complaint: things that must be closed, released, or restored, even when the code in between raises.
@contextmanager turns a generator into a with block. Everything before yield is setup, everything after is teardown, and the finally is what makes the teardown unconditional.
from contextlib import contextmanager
@contextmanager
def managed(name):
"""Acquire something, hand it over, and always release it."""
print(f"open {name}")
resource = {"name": name, "open": True}
try:
yield resource
finally:
resource["open"] = False
print(f"close {name}")
with managed("database") as db:
print(" using", db["name"])
print()
try:
with managed("file") as handle:
print(" using", handle["name"])
raise ValueError("something went wrong")
except ValueError as err:
print("caught:", err)
open database using database close database open file using file close file caught: something went wrong
Step 5: Timing, and why it needs a home
The obvious helper, saved for last because it has a catch. Timing is measurement, so the value changes every run, which means a decorator that prints the duration produces output you cannot test against.
The fix is to have it record rather than print. Store the measurements somewhere the caller can inspect, and it becomes both testable and far more useful.
import time
from functools import wraps
TIMINGS = {}
def timer(func):
"""Record how long each call took, rather than printing it."""
@wraps(func)
def wrapper(*args, **kwargs):
started = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
TIMINGS.setdefault(func.__name__, []).append(time.perf_counter() - started)
return wrapper
@timer
def summarize(rows):
return sum(rows) / len(rows)
print("average:", summarize([2, 4, 6]))
print("average:", summarize([10, 20]))
runs = TIMINGS["summarize"]
print("calls recorded:", len(runs))
print("all measured as numbers:", all(isinstance(t, float) for t in runs))
print("all non-negative:", all(t >= 0 for t in runs))
average: 4.0 average: 15.0 calls recorded: 2 all measured as numbers: True all non-negative: True
The finished toolkit
All four helpers in one file. Save it as helpers.py next to your project and import what you need.
"""Small helpers worth having in every project."""
import logging
import time
from contextlib import contextmanager
from functools import lru_cache, wraps
log = logging.getLogger(__name__)
TIMINGS = {}
__all__ = ["timer", "retry", "cached", "managed", "TIMINGS"]
def timer(func):
"""Record each call's duration in TIMINGS[func.__name__]."""
@wraps(func)
def wrapper(*args, **kwargs):
started = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
TIMINGS.setdefault(func.__name__, []).append(time.perf_counter() - started)
return wrapper
def retry(attempts=3, exceptions=(Exception,), sleep=time.sleep):
"""Retry on `exceptions`, backing off 1s, 2s, 4s. Re-raises on the last try."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as err:
if attempt == attempts:
log.error("%s failed after %d attempts: %s",
func.__name__, attempts, err)
raise
log.warning("%s failed (%s); retrying", func.__name__, err)
if sleep:
sleep(2 ** (attempt - 1))
return wrapper
return decorator
def cached(maxsize=128):
"""lru_cache, named for what it is. Only ever put this on a pure function."""
return lru_cache(maxsize=maxsize)
@contextmanager
def managed(resource, close="close"):
"""Yield `resource`, then call its close method whatever happens."""
try:
yield resource
finally:
closer = getattr(resource, close, None)
if callable(closer):
closer()
# Import what you need: # from helpers import retry, cached, timer, managed # # @retry(attempts=3, exceptions=(ConnectionError,)) # def fetch(url): ... # # Nothing here prints; the helpers log instead, so they stay quiet # inside whatever program imports them.
Keep going, make it your own
The toolkit is the point, so grow it with the things you personally keep rewriting.
Add a @deprecated marker
Warn once when an old function is called. The warnings module is built for this, and unlike a log line it can be turned into an error in a test suite so you find every caller.
import warnings
def deprecated(reason):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
warnings.warn(f"{func.__name__} is deprecated: {reason}",
DeprecationWarning, stacklevel=2)
return func(*args, **kwargs)
return wrapper
return decorator
Make @timer log a slow call
Recording is good; being told is better. Give the decorator a threshold and have it log a warning only when a call is slower than you expected: the quiet version of performance monitoring.
def timer(func, slow_after=1.0):
@wraps(func)
def wrapper(*args, **kwargs):
started = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
taken = time.perf_counter() - started
if taken > slow_after:
log.warning("%s took %.2fs", func.__name__, taken)
return wrapper
Write a context manager that times a block
Not everything worth timing is a function. with timed("load"): measures an arbitrary block, and it is four lines because you already know both halves.
@contextmanager
def timed(label):
started = time.perf_counter()
try:
yield
finally:
TIMINGS.setdefault(label, []).append(time.perf_counter() - started)
Test the toolkit
Helpers are exactly the kind of code worth testing: small, pure, and used everywhere, so a bug here shows up in every project at once. Pass a fake sleep and the retry tests run instantly.
def test_retry_gives_up_and_reraises():
calls = []
@retry(attempts=2, sleep=None)
def always_fails():
calls.append(1)
raise ConnectionError("nope")
try:
always_fails()
except ConnectionError:
pass
assert len(calls) == 2, "should try exactly twice"
Download the files
The finished module and a demo that exercises every helper. Run python demo.py to watch the retry, the cache, and the cleanup do their jobs.
- helpers.py (the finished toolkit) timer, retry, cached and managed, ready to drop into your next project.
- demo.py (exercises every helper) A runnable demonstration, including the failure cases.
Mini exercise (hard)
Write @retry yourself. The three-layer decorator is the thing worth being able to produce from memory. It should retry only the exception types it was given, re-raise on the final attempt, and record each attempt so the test can see what happened.
Write it out. Finish the snippet below and press Run. It works straight away in your browser.
from functools import wraps
log = []
def retry(attempts=3, exceptions=(Exception,)):
"""Retry on `exceptions` only; re-raise on the final attempt."""
def decorator(func):
# TODO: return a wrapper that retries
return func
return decorator
@retry(attempts=3, exceptions=(ConnectionError,))
def flaky():
log.append("try")
if len(log) < 3:
raise ConnectionError("reset")
return "ok"
@retry(attempts=3, exceptions=(ConnectionError,))
def broken():
log.append("bad")
raise ValueError("a real bug")
print("flaky:", flaky(), "after", len(log), "attempts")
log.clear()
try:
broken()
except ValueError as err:
print("broken raised immediately after", len(log), "attempt:", err)
Three nested functions: retry(attempts, exceptions) returns decorator(func), which returns wrapper(*args, **kwargs). Inside the wrapper, loop for attempt in range(1, attempts + 1), return func(...) inside a try, and in the except re-raise when attempt == attempts.
from functools import wraps
log = []
def retry(attempts=3, exceptions=(Exception,)):
"""Retry on `exceptions` only; re-raise on the final attempt."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exceptions:
if attempt == attempts:
raise
return wrapper
return decorator
@retry(attempts=3, exceptions=(ConnectionError,))
def flaky():
log.append("try")
if len(log) < 3:
raise ConnectionError("reset")
return "ok"
@retry(attempts=3, exceptions=(ConnectionError,))
def broken():
log.append("bad")
raise ValueError("a real bug")
print("flaky:", flaky(), "after", len(log), "attempts")
log.clear()
try:
broken()
except ValueError as err:
print("broken raised immediately after", len(log), "attempt:", err)
flaky: ok after 3 attempts
broken raised immediately after 1 attempt: a real bug
log.clear()
assert flaky() == "ok", "the third attempt should succeed and its value returned"
assert len(log) == 3, "it should have taken exactly three attempts"
log.clear()
try:
broken()
except ValueError:
pass
else:
raise AssertionError("a ValueError is not in `exceptions`, so it must not be retried or swallowed")
assert len(log) == 1, "an exception type you did not ask to retry must fail on the first attempt"
assert flaky.__name__ == "flaky", "use @wraps so the function keeps its own name"
print("✓ Retry behaves!")