Unit 10 Project: A Retry-and-Timer Toolkit
Put the unit in one box and use it
In this lesson
Every lesson in this unit answered a complaint about code you had already written. This project collects the answers into one small module and points it at the script from Units 7 to 9, the one that fetches records, cleans them, and stores them. The script's logic does not change. What changes is how much repetition surrounds it.
Explain it like I’m 5
You have been making tools all unit. Now you put them in one box and use them on a real job.
What goes in the box
Three pieces, each replacing something Unit 9 left duplicated:
@retry— the attempt loop, written once instead of around every network call.@timer— timing that records rather than prints, so it can go through the Unit 9 logger.run_section— a context manager wrapping a whole stage, so its cleanup happens even when the stage fails.
Note what is not here. There is no custom iterator: the file streaming is a generator, which is shorter and does the same job. Leaving a tool out because it is not needed is part of the exercise.
Assembling it
Everything below is the unit's material with nothing new: a decorator with arguments, a decorator that records, and a generator-based context manager.
import functools, time
from contextlib import contextmanager
DURATIONS = {}
def timer(func):
"""Record how long each call took, under the function's name."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
DURATIONS[func.__name__] = time.perf_counter() - start
return wrapper
def retry(times):
"""Try again on failure, up to `times` attempts."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ValueError:
if attempt == times:
raise
return wrapper
return decorator
@contextmanager
def run_section(name):
"""Announce a stage, and always announce its end."""
print(f"[{name}] start")
try:
yield
finally:
print(f"[{name}] end")
Use the toolkit on a flaky job. fetch fails twice before succeeding. Decorate it so it retries up to three times and records its duration, then run it inside a run_section. The decorators and context manager are supplied; you write the three lines that apply them.
import functools, time
from contextlib import contextmanager
DURATIONS = {}
def timer(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
DURATIONS[func.__name__] = time.perf_counter() - start
return wrapper
def retry(times):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ValueError:
if attempt == times:
raise
return wrapper
return decorator
@contextmanager
def run_section(name):
print(f"[{name}] start")
try:
yield
finally:
print(f"[{name}] end")
calls = []
# TODO: decorate fetch so it is timed, and retries up to 3 times
def fetch():
calls.append(1)
if len(calls) < 3:
raise ValueError("flaky")
return "data"
# TODO: run fetch() inside a run_section called "load" and print the result
Stack them with @timer on top and @retry(times=3) underneath, so the timer measures all three attempts together. Then with run_section("load"): and print("got:", fetch()) inside it.
import functools, time
from contextlib import contextmanager
DURATIONS = {}
def timer(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
DURATIONS[func.__name__] = time.perf_counter() - start
return wrapper
def retry(times):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ValueError:
if attempt == times:
raise
return wrapper
return decorator
@contextmanager
def run_section(name):
print(f"[{name}] start")
try:
yield
finally:
print(f"[{name}] end")
calls = []
@timer
@retry(times=3)
def fetch():
calls.append(1)
if len(calls) < 3:
raise ValueError("flaky")
return "data"
with run_section("load"):
print("got:", fetch())
print("attempts:", len(calls))
print("timed:", "fetch" in DURATIONS)Knowing which piece earned its place
Finish by taking the toolkit away again, mentally. For each tool, ask: how many places use it? If the answer is one, the plain version was better and you have added indirection for nothing.
On a real script the usual outcome is that @retry and the context manager earn their keep because the same wrapping applies in several places, while a decorator you wrote for a single function does not. That judgment is the actual skill this unit is teaching — the syntax is the easy part.
Common mistake: Stacking the decorators in the order that hides the retry
Decorators apply bottom-up, which is the opposite of how the lines read.
Decide what you want to measure. @timer outermost times every attempt together; innermost times each attempt on its own.
Common mistake: Timing inside the retry and reporting a misleading duration
Both wrap the same function, so the difference is easy to miss.
If the timer is inside the retry, it records only the last attempt and hides how long the whole thing really took.
Common mistake: Building all four tools when the script needed two
The unit taught four things, so all four feel like deliverables.
Write the ones with more than one caller. An unused abstraction is code to maintain for no benefit.
Common mistake: Letting the retry swallow a failure by returning None
The loop ends and the wrapper falls off the bottom with nothing to return.
Re-raise on the final attempt, as the example does, so exhausting the retries is a real failure rather than a silent None.
Which tool removes the attempt loop repeated at every call site?
@retry is written once and applied with one line wherever a flaky call happens.
With @timer above @retry(times=3), what does the timer measure?
Decorators apply bottom-up, so retry wraps the function and timer wraps retry: the timer sees the whole sequence.
Why does the toolkit use a generator rather than a custom iterator class?
A custom iterator would be several methods for what yield does in a line or two.
Mini exercise (hard)
Extend the toolkit with a @log_calls decorator that records each call as a (name, args) pair in a module-level list, then apply all three tools to one function. Print the recorded calls, confirm the retry ran, and confirm a duration was captured.
Give it a shot. Complete the code and press Run; there’s nothing to download or configure.
import functools
LOGGED = []
def log_calls(func):
# TODO: record (name, args) in LOGGED, then call through
return func
def retry(times):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ValueError:
if attempt == times:
raise
return wrapper
return decorator
calls = []
# TODO: stack log_calls ABOVE retry so it logs one call per operation
def fetch(source):
calls.append(1)
if len(calls) < 2:
raise ValueError("flaky")
return f"data from {source}"
print(fetch("api"))
print("logged:", LOGGED)
print("attempts:", len(calls))Model log_calls on timer: @functools.wraps, accept *args, **kwargs, append the record, then return the call through. Stack it above @retry so it sees the whole operation once rather than once per attempt.
import functools
LOGGED = []
def log_calls(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
LOGGED.append((func.__name__, args))
return func(*args, **kwargs)
return wrapper
def retry(times):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ValueError:
if attempt == times:
raise
return wrapper
return decorator
calls = []
@log_calls
@retry(times=3)
def fetch(source):
calls.append(1)
if len(calls) < 2:
raise ValueError("flaky")
return f"data from {source}"
print(fetch("api"))
print("logged:", LOGGED)
print("attempts:", len(calls))