Custom Decorator Libraries
Reviewed & published by Brayan K
Design reusable decorator libraries that power frameworks, APIs, and production systems
Part of the free Python course at LearnCodingFast — hands-on lessons with examples you run in your browser, plus practice exercises and a quick quiz.
Why Build Your Own Decorator Library?
High-level Python systems need common behaviors that decorators provide:
- Logging — Track function calls and results
- Caching — Store expensive computation results
- Error handling & retries — Recover from failures
- Validation — Enforce input/output constraints
- Timing — Measure performance
- Access control — Enforce permissions
- Registration — Build plugin systems
A centralized decorator library provides tested utilities, ensures consistent behavior, and speeds up development.
The Core Pattern
Every reusable decorator follows this pattern. Always use functools.wraps to preserve metadata.
import functools
def decorator_template(func):
"""Template for creating decorators."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# Before: code that runs before the function
print(f"Calling {func.__name__}")
# Execute the original function
result = func(*args, **kwargs)
# After: code that runs after the function
print(f"Finished {func.__name__}")
return result
return wrapper
@decorator_template
def greet(name):
"""Greet someone."""
return f"Hello, {name}!"
# Test it
result = greet("Alice")
print(result)
# functools.wraps preserves:
print(f"\nFunction name: {greet.__name__}")
print(f"Docstring: {greet.__doc__}")
# Without functools.wraps, name would be 'wrapper'!
# ✅ Expected output:
# Calling greet
# Finished greet
# Hello, Alice!
#
# Function name: greet
# Docstring: Greet someone.Worked Example: A Decorator, Line by Line
Before you build a library of them, make sure the machinery is completely clear. Read this program top to bottom — every non-obvious line says what it produces and why. Run it, then change something small (delete the @functools.wraps line) and run it again to see what breaks.
import functools
# A decorator is just a function that TAKES a function and RETURNS a new one.
def announce(func):
# functools.wraps copies func's __name__ and __doc__ onto wrapper,
# so the decorated function still LOOKS like the original one.
@functools.wraps(func)
def wrapper(*args, **kwargs): # *args/**kwargs = accept ANY arguments
print(f"-> starting {func.__name__}") # runs BEFORE the real function
result = func(*args, **kwargs) # call the real function, keep its return value
print(f"<- finished {func.__name__}") # runs AFTER the real function
return result # hand the real result back to the caller
return wrapper # the name 'add' will now point at wrapper
@announce # this line is exactly the same as writing: add = announce(add)
def add(a, b):
"""Add two numbers."""
return a + b
total = add(2, 3) # prints the two arrows, then gives back 5
print("total =", total)
# Because of functools.wraps, the original identity survives:
print("name:", add.__name__) # 'add' (without wraps this would say 'wrapper')
print("doc :", add.__doc__) # 'Add two numbers.'
# ✅ Expected output:
# -> starting add
# <- finished add
# total = 5
# name: add
# doc : Add two numbers.Reusable Timer Decorator
A practical decorator for measuring function execution time.
import functools
import time
def timing(func):
"""Measure and print function execution time."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
duration = time.perf_counter() - start
print(f"{func.__name__} took {duration:.4f}s")
return result
return wrapper
# Example usage
@timing
def calculate_sum(n):
"""Calculate sum of numbers from 1 to n."""
return sum(range(n + 1))
@timing
def slow_operation():
"""Simulate a slow operation."""
time.sleep(0.5)
return "Done"
# Test
print(f"Sum: {calculate_sum(1000000)}")
slow_operation()
# Great for debugging performance during development!Decorators With Arguments (Decorator Factories)
To accept arguments, wrap your decorator in another function that returns the actual decorator.
import functools
import time
def retry(times=3, delay=0):
"""
Retry a function if it fails.
Args:
times: Number of retry attempts
delay: Seconds to wait between retries
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
print(f"Attempt {attempt} failed: {e}")
if attempt < times:
time.sleep(delay)
# If all retries failed, raise the last exception
raise last_exception
return wrapper
return decorator
# Example: flaky network request
call_count = 0
@retry(times=5, delay=0.2)
def unreliable_api():
"""Simulates an unreliable API call."""
global call_count
call_count += 1
if call_count < 3:
raise ConnectionError("Network error")
return {"status": "success", "data": "Hello"}
# Test
print("Calling unreliable API:")
result = unreliable_api()
print(f"Result: {result}")
# Used in: API wrappers, database connections, cloud clients
# ✅ Expected output:
# Calling unreliable API:
# Attempt 1 failed: Network error
# Attempt 2 failed: Network error
# Result: {'status': 'success', 'data': 'Hello'}🎯 Your Turn: Finish the @repeat(n) Factory
Everything below is written for you except the three lines that make a decorator factory work. Fill in each ___, run it, and compare with the expected output at the bottom of the snippet. Remember the three layers: the outer function takes the argument, the middle one takes the function, and the inner one takes the call.
import functools
# 🎯 YOUR TURN — replace each ___ with the missing code.
# Goal: finish 'repeat', a decorator factory that runs a function n times
# and returns a list holding every result.
def repeat(times): # LAYER 1 — takes the argument (how many times)
def decorator(func): # LAYER 2 — takes the function being decorated
___ # 👉 add the line that preserves func's name and docstring
def wrapper(*args, **kwargs): # LAYER 3 — takes the actual call
results = []
for _ in range(times):
results.append(func(*args, **kwargs)) # call the real function each time
return results
return wrapper # layer 2 hands back the replacement function
return ___ # 👉 layer 1 must hand back layer 2
@___ # 👉 use the factory so shout() runs 3 times
def shout(word):
"""Return the word in capitals."""
return word.upper() + "!"
print(shout("hello"))
print("name:", shout.__name__)
# ✅ Expected output:
# ['HELLO!', 'HELLO!', 'HELLO!']
# name: shoutReusable Logging Decorator
Log function calls with customizable logger for debugging and monitoring.
import functools
def log_calls(logger=print):
"""
Log function calls with arguments and return values.
Args:
logger: Callable for logging (default: print)
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# Format arguments
args_str = ", ".join(repr(a) for a in args)
kwargs_str = ", ".join(f"{k}={v!r}" for k, v in kwargs.items())
all_args = ", ".join(filter(None, [args_str, kwargs_str]))
logger(f"→ Calling {func.__name__}({all_args})")
try:
result = func(*args, **kwargs)
logger(f"← {func.__name__} returned {result!r}")
return result
except Exception as e:
logger(f"✗ {func.__name__} raised {type(e).__name__}: {e}")
raise
return wrapper
return decorator
# Example with print
@log_calls()
def add(a, b):
return a + b
@log_calls()
def divide(a, b):
return a / b
# Test
print("Example 1: Success")
add(5, 3)
print("\nExample 2: Error handling")
try:
divide(10, 0)
except ZeroDivisionError:
pass
# Can be used with real logging
import logging
log = logging.getLogger("app")
@log_calls(log.info)
def process_data(data):
return f"Processed: {data}"
# ✅ Expected output:
# Example 1: Success
# → Calling add(5, 3)
# ← add returned 8
#
# Example 2: Error handling
# → Calling divide(10, 0)
# ✗ divide raised ZeroDivisionError: division by zeroType Validation Decorator
Enforce type checking using function annotations for safer code.
import functools
import inspect
def enforce_types(func):
"""Validate argument types based on annotations."""
sig = inspect.signature(func)
@functools.wraps(func)
def wrapper(*args, **kwargs):
# Bind arguments to parameters
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
# Check each parameter
for param_name, param_value in bound.arguments.items():
param = sig.parameters[param_name]
# Skip if no annotation
if param.annotation is inspect.Parameter.empty:
continue
expected_type = param.annotation
# Validate type
if not isinstance(param_value, expected_type):
raise TypeError(
f"Argument '{param_name}' must be {expected_type.__name__}, "
f"got {type(param_value).__name__}"
)
return func(*args, **kwargs)
return wrapper
# Example usage
@enforce_types
def create_user(name: str, age: int, active: bool = True):
"""Create a user with validated types."""
return {"name": name, "age": age, "active": active}
# Test valid calls
print("Valid calls:")
print(create_user("Alice", 30))
print(create_user("Bob", 25, False))
# Test invalid calls
print("\nInvalid calls:")
try:
create_user("Charlie", "not a number")
except TypeError as e:
print(f"Error: {e}")
try:
create_user(123, 30) # name should be str
except TypeError as e:
print(f"Error: {e}")
# Great for: API endpoints, CLI tools, data pipelines
# ✅ Expected output:
# Valid calls:
# {'name': 'Alice', 'age': 30, 'active': True}
# {'name': 'Bob', 'age': 25, 'active': False}
#
# Invalid calls:
# Error: Argument 'age' must be int, got str
# Error: Argument 'name' must be str, got intAccess Control Decorator
Enforce role-based access control, commonly used in web frameworks and APIs.
import functools
def require_role(required_role):
"""
Require a specific role to execute function.
Args:
required_role: Role required to access the function
"""
def decorator(func):
@functools.wraps(func)
def wrapper(user, *args, **kwargs):
if not hasattr(user, 'role'):
raise AttributeError("User object must have 'role' attribute")
if user.role != required_role:
raise PermissionError(
f"Access denied. Required role: {required_role}, "
f"user has: {user.role}"
)
return func(user, *args, **kwargs)
return wrapper
return decorator
# Simple user class
class User:
def __init__(self, name, role):
self.name = name
self.role = role
# Protected functions
@require_role("admin")
def delete_user(user, target_id):
return f"{user.name} deleted user {target_id}"
@require_role("moderator")
def ban_user(user, target_id):
return f"{user.name} banned user {target_id}"
@require_role("user")
def view_profile(user):
return f"{user.name} viewing profile"
# Test
admin = User("Alice", "admin")
moderator = User("Bob", "moderator")
regular = User("Charlie", "user")
print("Admin actions:")
print(delete_user(admin, 123))
print(view_profile(admin))
print("\nModerator actions:")
print(ban_user(moderator, 456))
print("\nRegular user trying admin action:")
try:
delete_user(regular, 789)
except PermissionError as e:
print(f"Error: {e}")
# Used in: web frameworks, APIs, internal toolsMemoization / Caching Decorator
Cache expensive function results to improve performance.
import functools
def memoize(func):
"""Cache function results based on arguments."""
cache = {}
@functools.wraps(func)
def wrapper(*args):
if args not in cache:
print(f"Computing {func.__name__}{args}...")
cache[args] = func(*args)
else:
print(f"Using cached result for {func.__name__}{args}")
return cache[args]
# Expose cache for inspection
wrapper.cache = cache
return wrapper
# Example: expensive calculation
@memoize
def fibonacci(n):
"""Calculate fibonacci number (slow recursive version)."""
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print("First call - computes:")
print(f"fib(10) = {fibonacci(10)}")
print("\nSecond call - cached:")
print(f"fib(10) = {fibonacci(10)}")
print("\nCache contents:")
print(f"Cache size: {len(fibonacci.cache)}")
# Better version: functools.lru_cache
@functools.lru_cache(maxsize=128)
def fibonacci_fast(n):
"""Faster version using built-in LRU cache."""
if n < 2:
return n
return fibonacci_fast(n - 1) + fibonacci_fast(n - 2)
print("\nUsing LRU cache:")
print(f"fib(35) = {fibonacci_fast(35)}")
print(f"Cache info: {fibonacci_fast.cache_info()}")
# Used in: expensive calculations, API calls, ML preprocessing
# ✅ Expected output:
# First call - computes:
# Computing fibonacci(10,)...
# Computing fibonacci(9,)...
# Computing fibonacci(8,)...
# Computing fibonacci(7,)...
# Computing fibonacci(6,)...
# Computing fibonacci(5,)...
# Computing fibonacci(4,)...
# Computing fibonacci(3,)...
# Computing fibonacci(2,)...
# Computing fibonacci(1,)...
# Computing fibonacci(0,)...
# Using cached result for fibonacci(1,)
# Using cached result for fibonacci(2,)
# Using cached result for fibonacci(3,)
# Using cached result for fibonacci(4,)
# Using cached result for fibonacci(5,)
# Using cached result for fibonacci(6,)
# Using cached result for fibonacci(7,)
# Using cached result for fibonacci(8,)
# fib(10) = 55
#
# Second call - cached:
# Using cached result for fibonacci(10,)
# fib(10) = 55
#
# Cache contents:
# Cache size: 11
#
# Using LRU cache:
# fib(35) = 9227465
# Cache info: CacheInfo(hits=33, misses=36, maxsize=128, currsize=36)Registry Pattern for Plugins
Automatically register functions or classes for plugin systems and routing.
import functools
# Global registry
command_registry = {}
def register_command(name):
"""
Register a function as a command handler.
Args:
name: Command name for the registry
"""
def decorator(func):
command_registry[name] = func
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
# Register handlers
@register_command("greet")
def handle_greet(name):
"""Greet a user."""
return f"Hello, {name}!"
@register_command("calculate")
def handle_calculate(a, b, operation="add"):
"""Perform a calculation."""
ops = {
"add": a + b,
"subtract": a - b,
"multiply": a * b
}
return ops.get(operation, "Unknown operation")
@register_command("status")
def handle_status():
"""Get system status."""
return "System is running"
# List registered commands
print("Registered commands:")
for cmd_name in command_registry:
print(f" - {cmd_name}")
# Execute commands dynamically
print("\nExecuting commands:")
def execute_command(command_name, *args, **kwargs):
"""Execute a registered command."""
if command_name in command_registry:
handler = command_registry[command_name]
return handler(*args, **kwargs)
return f"Unknown command: {command_name}"
print(execute_command("greet", "Alice"))
print(execute_command("calculate", 10, 5, operation="multiply"))
print(execute_command("status"))
# This pattern is used in:
# - Flask/FastAPI routing
# - CLI command handlers
# - Plugin systems
# - Event handlers
# ✅ Expected output:
# Registered commands:
# - greet
# - calculate
# - status
#
# Executing commands:
# Hello, Alice!
# 50
# System is runningAsync-Compatible Decorators
Support both sync and async functions in your decorator library.
import functools
import inspect
import time
def universal_timer(func):
"""Timer that works with both sync and async functions."""
if inspect.iscoroutinefunction(func):
# Async version
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
start = time.perf_counter()
result = await func(*args, **kwargs)
duration = time.perf_counter() - start
print(f"{func.__name__} took {duration:.4f}s (async)")
return result
return async_wrapper
else:
# Sync version
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
duration = time.perf_counter() - start
print(f"{func.__name__} took {duration:.4f}s (sync)")
return result
return sync_wrapper
# Sync function
@universal_timer
def sync_task():
time.sleep(0.1)
return "Sync done"
# Async function
@universal_timer
async def async_task():
# Would use 'await asyncio.sleep(0.1)' in real async code
return "Async done"
# Test sync
print("Sync test:")
result = sync_task()
print(f"Result: {result}")
# To test async (note: can't actually await here)
print("\nAsync test:")
print("In real async code, you would: await async_task()")
# Inspect what was created
print(f"\nIs sync_task a coroutine? {inspect.iscoroutinefunction(sync_task)}")
print(f"Is async_task a coroutine? {inspect.iscoroutinefunction(async_task)}")
# This pattern is essential for:
# - FastAPI route handlers
# - aiohttp middleware
# - async database operations
# - websocket handlersStacking Decorators Properly
Multiple decorators execute bottom-to-top, but wrap top-to-bottom. Order matters!
import functools
# Three simple decorators
def decorator_a(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("A: Before")
result = func(*args, **kwargs)
print("A: After")
return result
return wrapper
def decorator_b(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("B: Before")
result = func(*args, **kwargs)
print("B: After")
return result
return wrapper
def decorator_c(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("C: Before")
result = func(*args, **kwargs)
print("C: After")
return result
return wrapper
# Stack them
@decorator_a # Wraps last (outermost)
@decorator_b # Wraps second
@decorator_c # Wraps first (innermost)
def greet(name):
print(f"Hello, {name}!")
return "done"
print("Execution order:")
greet("Alice")
print("\n" + "="*50)
print("Decorator stacking rules:")
print("="*50)
print("Wrapping: A wraps B wraps C wraps original")
print("Execution: A before → B before → C before → original")
print(" → C after → B after → A after")
# Practical example: proper ordering
def authenticate(func):
"""Must run first - verify user."""
@functools.wraps(func)
def wrapper(user, *args, **kwargs):
if not user:
raise ValueError("Authentication required")
print(f"✓ Authenticated: {user}")
return func(user, *args, **kwargs)
return wrapper
def validate_input(func):
"""Runs after auth - check inputs."""
@functools.wraps(func)
def wrapper(user, data, *args, **kwargs):
if not data:
raise ValueError("Data required")
print(f"✓ Validated: {data}")
return func(user, data, *args, **kwargs)
return wrapper
def log_action(func):
"""Runs last - log everything."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"→ Logging: {func.__name__}")
result = func(*args, **kwargs)
print(f"← Logged: {result}")
return result
return wrapper
@log_action # Outer: logs everything
@validate_input # Middle: validates data
@authenticate # Inner: checks auth first
def process_data(user, data):
return f"Processed {data} for {user}"
print("\nPractical stacking:")
result = process_data("Alice", "file.txt")
print(f"\nFinal result: {result}")
# ✅ Expected output:
# Execution order:
# A: Before
# B: Before
# C: Before
# Hello, Alice!
# C: After
# B: After
# A: After
#
# ==================================================
# Decorator stacking rules:
# ==================================================
# Wrapping: A wraps B wraps C wraps original
# Execution: A before → B before → C before → original
# → C after → B after → A after
#
# Practical stacking:
# → Logging: process_data
# ✓ Validated: file.txt
# ✓ Authenticated: Alice
# ← Logged: Processed file.txt for Alice
#
# Final result: Processed file.txt for AliceBuilding a Complete Decorator Library
Organize decorators into a reusable library structure for production use.
Advanced: Exponential Backoff Retry
Production-grade retry decorator with exponential backoff and jitter.
Summary
You've learned to build production-grade decorator libraries:
- Core decorator pattern with functools.wraps
- Decorator factories for parameterized decorators
- Timing and performance measurement
- Logging with customizable loggers
- Type validation using annotations
- Access control and permissions
- Memoization and caching
- Registry patterns for plugins
- Async-compatible decorators
- Proper decorator stacking
- Exponential backoff retry
- Library organization and best practices
Decorator libraries power major frameworks like Django, FastAPI, Flask, and Celery. A well-designed decorator library becomes reusable infrastructure that speeds up development across multiple projects.
📋 Quick Reference — Decorator Libraries
| Pattern | Use case |
|---|---|
| @functools.wraps(fn) | Preserve original __name__ and __doc__ |
| def decorator(fn): ... | Basic decorator factory |
| def with_args(n): ... | Decorator that accepts arguments |
| class-based decorator | Decorator with persistent state |
| @retry(max=3, delay=1) | Parameterised production decorator |
🎯 Mini-Challenge: @count_calls
Now write one from scratch. Your decorator should remember how many times the function it wraps has been called, and expose that number to the outside world. The trick is storing the counter somewhere that survives between calls — a plain local variable will not, but an attribute on the wrapper function itself will.
🎉 Great work! You've completed this lesson.
You can now build reusable decorator libraries — retry logic, rate limiting, caching, auth guards — used across entire codebases.
Practice quiz
Why should every reusable decorator apply @functools.wraps(func) to its wrapper?
- It makes the wrapper run faster
- It automatically caches the result
- It preserves the original function's __name__ and __doc__
- It is required for the decorator to work at all
Answer: It preserves the original function's __name__ and __doc__. Without functools.wraps the wrapped function's __name__ would become 'wrapper'; wraps copies over the original name and docstring.
How do you write a decorator that accepts arguments, like @retry(times=5)?
- Wrap the decorator in an outer function that returns the actual decorator (a decorator factory)
- Add the arguments directly to wrapper
- Use @functools.wraps with arguments
- It is impossible in Python
Answer: Wrap the decorator in an outer function that returns the actual decorator (a decorator factory). A decorator factory is an outer function taking the arguments and returning the real decorator, which returns the wrapper.
In the lesson's memoize decorator, after calling fibonacci(10) the cache holds how many entries?
- 1
- 10
- 55
- 11
Answer: 11. The recursive fibonacci(10) computes and caches results for n = 0 through 10, which is 11 distinct entries.
What does functools.lru_cache(maxsize=128) provide over a hand-written memoize?
- Nothing, they are identical
- A built-in, size-bounded cache with cache_info() statistics
- Automatic async support
- Type validation of arguments
Answer: A built-in, size-bounded cache with cache_info() statistics. lru_cache is a tested standard-library cache with a max size and a cache_info() method reporting hits and misses.
When decorators are stacked, in what order do they execute around the call?
- The innermost (bottom) wraps first, and the outermost (top) runs first around the call
- Top-to-bottom both wrapping and executing
- Random order
- Only the top decorator runs
Answer: The innermost (bottom) wraps first, and the outermost (top) runs first around the call. The decorator closest to the function wraps first, but the outermost decorator's 'before' code runs first when the function is called.
How does the universal_timer decorator support both sync and async functions?
- It uses two threads
- It only works on sync functions
- It checks inspect.iscoroutinefunction(func) and returns an async or sync wrapper accordingly
- It converts async functions to sync
Answer: It checks inspect.iscoroutinefunction(func) and returns an async or sync wrapper accordingly. inspect.iscoroutinefunction(func) detects coroutines so the decorator can return an async wrapper that awaits, or a plain sync wrapper.
What is the purpose of the registry pattern shown with @register_command('greet')?
- To cache return values
- To automatically register functions in a lookup so they can be dispatched dynamically
- To enforce type annotations
- To time function execution
Answer: To automatically register functions in a lookup so they can be dispatched dynamically. The decorator stores each function in a registry dict keyed by name, enabling dynamic command dispatch like routing or plugins.
The enforce_types decorator validates argument types using what?
- Random sampling
- A hardcoded list of types
- The __doc__ string
- The function's annotations read via inspect.signature
Answer: The function's annotations read via inspect.signature. It builds inspect.signature(func), binds the call's arguments, and compares each value against the parameter's annotation.
Why does the require_role access-control decorator raise PermissionError?
- When functools.wraps is missing
- When the user's role does not match the required_role
- When the function is async
- When the cache is empty
Answer: When the user's role does not match the required_role. If user.role differs from the required role the wrapper raises PermissionError, denying access.
What does exponential backoff with jitter prevent in the backoff_retry decorator?
- Memory leaks
- Type errors
- Many clients retrying in lockstep (the thundering-herd problem)
- Infinite recursion
Answer: Many clients retrying in lockstep (the thundering-herd problem). Increasing delays plus random jitter spread out retries so failed clients don't all hammer the service at the same instant.
Continue this course
- Previous: Metaprogramming & Introspection with inspect
- Next: Using SQLite & ORMs (SQLAlchemy) in Python — Query databases with raw SQLite and the SQLAlchemy ORM
- Quick reference: Python cheat sheet
- From the blog: Python Decorators: A Practical Guide