Chapter 8 — Decorators and Cross-Cutting Behaviour
1. Opening problem
Several functions repeat logging:
def create_user(
email: str,
) -> None:
print("Starting create_user")
...
print("Finished create_user")
def delete_user(
user_id: int,
) -> None:
print("Starting delete_user")
...
print("Finished delete_user")
A decorator can add reusable behaviour:
@log_calls
def create_user(
email: str,
) -> None:
...
Decorators are used for:
- logging;
- retries;
- metrics;
- caching;
- authorization;
- transactions;
- validation;
- route registration;
- framework integration.
2. Manual decoration
Start with:
def greet(name: str) -> str:
return f"Hello, {name}"
Decorator:
def log_calls(function):
def wrapper(*args, **kwargs):
print(
f"Calling {function.__name__}"
)
result = function(
*args,
**kwargs,
)
print(
f"Finished {function.__name__}"
)
return result
return wrapper
Apply manually:
greet = log_calls(greet)
Decorator syntax means the same thing:
@log_calls
def greet(name: str) -> str:
return f"Hello, {name}"
At its core:
target = decorator(target)
3. Decoration time
def announce(function):
print(
f"Decorating {function.__name__}"
)
return function
@announce
def process() -> None:
print("Processing")
The decorator runs when the definition executes, normally during module import.
This matters because decorators may:
- register routes;
- access configuration;
- change classes;
- create global state;
- perform startup side effects.
Avoid expensive or unpredictable work during import.
4. Preserving metadata
Without help:
def log_calls(function):
def wrapper(*args, **kwargs):
return function(*args, **kwargs)
return wrapper
After decoration:
print(greet.__name__)
may print "wrapper".
Use functools.wraps:
from functools import wraps
def log_calls(function):
@wraps(function)
def wrapper(*args, **kwargs):
return function(*args, **kwargs)
return wrapper
wraps preserves:
- name;
- documentation;
- annotations;
- module;
__wrapped__.
This supports debugging, documentation, tests, frameworks, and introspection.
5. Typed decorators
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def log_calls(
function: Callable[P, R],
) -> Callable[P, R]:
@wraps(function)
def wrapper(
*args: P.args,
**kwargs: P.kwargs,
) -> R:
print(
f"Calling {function.__name__}"
)
result = function(
*args,
**kwargs,
)
print(
f"Finished {function.__name__}"
)
return result
return wrapper
ParamSpec preserves parameters. TypeVar preserves the return type.
6. Parameterized decorators
Usage:
@retry(max_attempts=3)
def load_data() -> bytes:
...
This has three levels:
- factory receives options;
- decorator receives function;
- wrapper receives call arguments.
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def retry(
*,
max_attempts: int,
) -> Callable[
[Callable[P, R]],
Callable[P, R],
]:
if max_attempts < 1:
raise ValueError(
"max_attempts must be positive"
)
def decorator(
function: Callable[P, R],
) -> Callable[P, R]:
@wraps(function)
def wrapper(
*args: P.args,
**kwargs: P.kwargs,
) -> R:
for attempt in range(
1,
max_attempts + 1,
):
try:
return function(
*args,
**kwargs,
)
except Exception:
if attempt == max_attempts:
raise
raise RuntimeError("Unreachable")
return wrapper
return decorator
This first version is dangerous because it retries every exception.
7. Selective retries
from time import sleep
def retry(
*,
max_attempts: int,
exceptions: tuple[
type[Exception],
...
],
delay_seconds: float = 0.0,
):
if max_attempts < 1:
raise ValueError(
"max_attempts must be positive"
)
if delay_seconds < 0:
raise ValueError(
"delay must not be negative"
)
def decorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
for attempt in range(
1,
max_attempts + 1,
):
try:
return function(
*args,
**kwargs,
)
except exceptions:
if attempt == max_attempts:
raise
if delay_seconds:
sleep(delay_seconds)
raise RuntimeError("Unreachable")
return wrapper
return decorator
Usage:
@retry(
max_attempts=3,
exceptions=(
TimeoutError,
ConnectionError,
),
delay_seconds=0.5,
)
def load_remote_data() -> bytes:
...
Production retry logic should consider:
- idempotency;
- exponential backoff;
- jitter;
- maximum elapsed time;
- cancellation;
- observability;
- which errors are transient.
8. Decorator order
@authorization_required
@log_calls
def delete_user(
user_id: int,
) -> None:
...
Equivalent:
delete_user = authorization_required(
log_calls(delete_user)
)
The nearest decorator wraps first. At call time, the outermost wrapper receives control first.
Order can change:
- authorization boundaries;
- cache visibility;
- transaction scope;
- retry transaction behaviour;
- measured duration;
- logged failures.
Decorator order is an architectural decision.
9. Methods
A general wrapper also receives self:
class UserService:
@log_calls
def create(
self,
email: str,
) -> int:
return 42
The wrapper receives (self, email) through *args.
Do not assume the first argument means self unless the decorator is intentionally method-specific.
10. Timing decorator
from time import perf_counter
def measure_time(
function: Callable[P, R],
) -> Callable[P, R]:
@wraps(function)
def wrapper(
*args: P.args,
**kwargs: P.kwargs,
) -> R:
started = perf_counter()
try:
return function(
*args,
**kwargs,
)
finally:
duration = (
perf_counter() - started
)
print(
f"{function.__name__} took "
f"{duration:.6f} seconds"
)
return wrapper
finally records duration even when an exception occurs.
Production code should emit structured metrics or logs rather than print.
11. Authorization decorator
class AuthorizationError(Exception):
pass
def requires_role(role: str):
def decorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
user = get_current_user()
if role not in user.roles:
raise AuthorizationError(
f"Required role: {role}"
)
return function(
*args,
**kwargs,
)
return wrapper
return decorator
Cautions:
- hidden context reduces test clarity;
- framework state may leak into domain code;
- broad access checks do not replace domain authorization;
- the caller may not see the dependency.
Use decorators at clear application boundaries. Keep business permission rules explicit.
12. Caching
from functools import lru_cache
@lru_cache(maxsize=128)
def calculate(value: int) -> int:
...
Caching is safe only when you understand:
- argument hashability;
- determinism;
- stale values;
- memory growth;
- shared mutable returns;
- authorization context;
- invalidation;
- side effects.
Never cache a side-effecting function merely because the decorator is easy to add.
13. Registration decorators
from collections.abc import Callable
Command = Callable[[], str]
COMMANDS: dict[str, Command] = {}
def command(name: str):
def decorator(
function: Command,
) -> Command:
if name in COMMANDS:
raise ValueError(
f"Duplicate command: {name}"
)
COMMANDS[name] = function
return function
return decorator
Usage:
@command("hello")
def hello_command() -> str:
return "Hello"
Registration happens during module import.
Common uses:
- web routes;
- CLI commands;
- events;
- plugins;
- test discovery.
Risks:
- import-order dependence;
- hidden global state;
- duplicate registration;
- test isolation problems.
14. Async decorators
A synchronous wrapper is not enough for an async function.
from collections.abc import Awaitable
def log_async_calls(
function: Callable[
P,
Awaitable[R],
],
) -> Callable[
P,
Awaitable[R],
]:
@wraps(function)
async def wrapper(
*args: P.args,
**kwargs: P.kwargs,
) -> R:
print(
f"Calling {function.__name__}"
)
try:
return await function(
*args,
**kwargs,
)
finally:
print(
f"Finished {function.__name__}"
)
return wrapper
The wrapper must await the original coroutine.
15. Testing decorated functions
With wraps, the original function is available through:
decorated.__wrapped__
Test:
- core business behaviour;
- wrapper behaviour;
- combined integration;
- metadata preservation;
- failure propagation;
- exact retry count;
- non-retryable errors;
- async cancellation where relevant.
Do not bypass decorators in every test. That would leave important behaviour unverified.
16. Decorators versus explicit composition
Decorator:
@transactional
@requires_role("admin")
def delete_user(
user_id: int,
) -> None:
...
Explicit:
def delete_user(
user_id: int,
*,
transaction_manager,
authorization_service,
) -> None:
authorization_service.require_role(
"admin"
)
with transaction_manager:
...
Decorators are attractive when:
- behaviour is reusable;
- conventions are established;
- the function contract remains stable;
- order is documented;
- hidden control flow is acceptable.
Explicit composition is often better when:
- behaviour is domain-specific;
- dependencies should remain visible;
- control flow is complex;
- debugging transparency matters;
- failure handling differs per operation.
17. TypeScript comparison
Python decorator semantics reduce to:
target = decorator(target)
Python decorators commonly apply to:
- free functions;
- methods;
- classes;
- properties;
- class methods;
- static methods.
The feature follows directly from first-class functions and class objects.
18. Common mistakes
- forgetting
wraps; - retrying every exception;
- ignoring order;
- import-time network or database work;
- hiding business rules;
- wrapping async functions synchronously;
- losing type information;
- caching side effects;
- swallowing exceptions;
- using decorators where explicit code would be clearer.
19. English vocabulary
| Term | Meaning |
|---|---|
| decorator | callable transforming a function or class |
| wrapper | function adding behaviour around another function |
| cross-cutting concern | behaviour used across many modules |
| metadata | information describing a function |
| registration | storing behaviour for later discovery |
| retry | repeating after a transient failure |
| idempotent | safe to repeat without unintended change |
| backoff | increasing delay between attempts |
| instrumentation | adding logs, metrics, or tracing |
| composition | combining separate behaviours |
Useful sentences:
- “The decorator preserves the original signature.”
- “Retrying every exception hides programming defects.”
- “Decorator order changes the transaction boundary.”
- “This rule is domain-specific and should remain explicit.”
- “The async wrapper must await the coroutine.”
- “Registration occurs during module import.”
20. Speaking task
Explain for eight minutes:
How decorators work from first principles.
Include manual reassignment, closures, wraps, factories, ordering, and async functions.
21. Writing task
Review a retry decorator that catches Exception, retries five times immediately, and logs with print. Write approximately 350 words.
22. Exercises
Exercise 1
Implement a typed decorator logging start, success, and failure.
Exercise 2
Implement a selective retry decorator.
Exercise 3
Create a named-parameter validator using inspect.signature.
Exercise 4
Create a command-registration decorator that rejects duplicates.
Exercise 5
Create an async timing decorator.
Exercise 6
Explain:
@retry(...)
@transactional
def process():
...
versus:
@transactional
@retry(...)
def process():
...
23. Complete solutions
Solution 1
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def log_outcome(
function: Callable[P, R],
) -> Callable[P, R]:
@wraps(function)
def wrapper(
*args: P.args,
**kwargs: P.kwargs,
) -> R:
print(
f"START {function.__name__}"
)
try:
result = function(
*args,
**kwargs,
)
except Exception as error:
print(
f"FAIL {function.__name__}: "
f"{type(error).__name__}"
)
raise
print(
f"SUCCESS {function.__name__}"
)
return result
return wrapper
Solution 2
from time import sleep
def retry(
*,
max_attempts: int,
exceptions: tuple[
type[Exception],
...
],
delay_seconds: float = 0.0,
):
if max_attempts < 1:
raise ValueError(
"max_attempts must be positive"
)
def decorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
for attempt in range(
1,
max_attempts + 1,
):
try:
return function(
*args,
**kwargs,
)
except exceptions:
if attempt == max_attempts:
raise
sleep(delay_seconds)
raise RuntimeError("Unreachable")
return wrapper
return decorator
Solution 3
from inspect import signature
def require_non_empty(
parameter_name: str,
):
def decorator(function):
function_signature = signature(
function
)
if (
parameter_name
not in function_signature.parameters
):
raise ValueError(
f"Unknown parameter: "
f"{parameter_name}"
)
@wraps(function)
def wrapper(*args, **kwargs):
bound = function_signature.bind(
*args,
**kwargs,
)
bound.apply_defaults()
value = bound.arguments[
parameter_name
]
if (
not isinstance(value, str)
or not value.strip()
):
raise ValueError(
f"{parameter_name} must "
f"be a non-empty string"
)
return function(
*args,
**kwargs,
)
return wrapper
return decorator
Solution 4
from collections.abc import Callable
Command = Callable[[], str]
COMMANDS: dict[str, Command] = {}
def command(name: str):
def decorator(
function: Command,
) -> Command:
if name in COMMANDS:
raise ValueError(
f"Command already registered: "
f"{name}"
)
COMMANDS[name] = function
return function
return decorator
Solution 5
from time import perf_counter
def measure_async_time(function):
@wraps(function)
async def wrapper(
*args,
**kwargs,
):
started = perf_counter()
try:
return await function(
*args,
**kwargs,
)
finally:
duration = (
perf_counter() - started
)
print(
f"{function.__name__} took "
f"{duration:.6f} seconds"
)
return wrapper
Solution 6
First:
process = retry(...)(
transactional(process)
)
Each retry invokes the transactional wrapper. This can create a fresh transaction per attempt.
Second:
process = transactional(
retry(...)(process)
)
One transaction can surround the complete retry loop. A failed first attempt may leave transaction state unsuitable for the next attempt.
The correct arrangement depends on transaction semantics, but separate transactions per retry are often safer.
24. Chapter checkpoint
You should now be able to explain:
- decorator reassignment;
- decoration timing;
- metadata preservation;
- typed wrappers;
- decorator factories;
- selective retries;
- ordering;
- async wrappers;
- decorator versus explicit composition.