Creating decorators that accept parameters opens up a whole new level of flexibility and reusability in Python programming. While basic decorators are powerful, parameterized decorators allow you to customize their behavior, making them adaptable to different scenarios without writing separate decorator functions. This comprehensive guide will show you how to create sophisticated decorators that accept parameters and arguments, making your code more modular and maintainable.
Table Of Contents
- Understanding Decorator Parameters vs Arguments
- The Three-Layer Structure
- Basic Parameterized Decorator Examples
- Advanced Parameterized Decorators
- Best Practices for Parameterized Decorators
- Common Patterns and Pitfalls
- Conclusion
Understanding Decorator Parameters vs Arguments
Before diving into implementation, it's important to understand the terminology:
- Parameters: The variables defined in the decorator function signature
- Arguments: The actual values passed to the decorator when it's applied
# Basic decorator (no parameters)
@simple_decorator
def my_function():
pass
# Parameterized decorator (with arguments)
@parameterized_decorator(timeout=30, retries=3)
def my_function():
pass
The Three-Layer Structure
Parameterized decorators follow a three-layer structure:
- Outer function: Accepts the decorator parameters
- Middle function: The actual decorator that receives the function
- Inner function: The wrapper that executes around the original function
from functools import wraps
def my_decorator(param1, param2='default'): # Outer: accepts parameters
def decorator(func): # Middle: receives the function
@wraps(func)
def wrapper(*args, **kwargs): # Inner: wraps the function
# Use param1 and param2 here
return func(*args, **kwargs)
return wrapper
return decorator
Basic Parameterized Decorator Examples
Retry Decorator with Configurable Attempts
import time
import random
from functools import wraps
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
"""Decorator that retries function execution with configurable parameters."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
result = func(*args, **kwargs)
if attempt > 0:
print(f"✓ {func.__name__} succeeded on attempt {attempt + 1}")
return result
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
print(f"✗ Attempt {attempt + 1}/{max_attempts} failed: {e}")
if delay > 0:
print(f"⏳ Waiting {delay}s before retry...")
time.sleep(delay)
else:
print(f"✗ All {max_attempts} attempts failed")
raise last_exception
# Store configuration for introspection
wrapper._retry_config = {
'max_attempts': max_attempts,
'delay': delay,
'exceptions': exceptions
}
return wrapper
return decorator
# Usage examples with different configurations
@retry(max_attempts=5, delay=0.5)
def unreliable_network_call():
"""Simulate unreliable network operation."""
if random.random() < 0.6: # 60% failure rate
raise ConnectionError("Network timeout")
return "Success!"
@retry(max_attempts=2, delay=0, exceptions=(ValueError, TypeError))
def validate_input(data):
"""Validate input data with specific error handling."""
if not isinstance(data, (int, float)):
raise TypeError("Data must be numeric")
if data < 0:
raise ValueError("Data must be positive")
return f"Valid data: {data}"
# Test the decorated functions
try:
result = unreliable_network_call()
print(f"Network result: {result}")
except ConnectionError as e:
print(f"Network failed: {e}")
try:
result = validate_input("invalid")
print(f"Validation result: {result}")
except (ValueError, TypeError) as e:
print(f"Validation failed: {e}")
# Access configuration
print(f"Network retry config: {unreliable_network_call._retry_config}")
Timing Decorator with Output Options
import time
from functools import wraps
from datetime import datetime
def timing(unit='seconds', precision=4, show_args=False, logger=None):
"""Decorator that measures execution time with flexible output options."""
# Conversion factors for different time units
unit_factors = {
'seconds': 1,
'milliseconds': 1000,
'microseconds': 1000000,
'nanoseconds': 1000000000
}
if unit not in unit_factors:
raise ValueError(f"Invalid unit: {unit}. Choose from {list(unit_factors.keys())}")
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.perf_counter()
try:
result = func(*args, **kwargs)
end_time = time.perf_counter()
execution_time = (end_time - start_time) * unit_factors[unit]
# Format the timing message
if show_args and (args or kwargs):
args_str = ', '.join(repr(arg) for arg in args)
kwargs_str = ', '.join(f'{k}={v!r}' for k, v in kwargs.items())
all_args = ', '.join(filter(None, [args_str, kwargs_str]))
message = f"{func.__qualname__}({all_args}) executed in {execution_time:.{precision}f} {unit}"
else:
message = f"{func.__qualname__} executed in {execution_time:.{precision}f} {unit}"
# Output the timing information
if logger:
logger.info(message)
else:
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"[{timestamp}] ⏱️ {message}")
return result
except Exception as e:
end_time = time.perf_counter()
execution_time = (end_time - start_time) * unit_factors[unit]
error_message = f"{func.__qualname__} failed after {execution_time:.{precision}f} {unit}: {e}"
if logger:
logger.error(error_message)
else:
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"[{timestamp}] ❌ {error_message}")
raise
# Add timing configuration
wrapper._timing_config = {
'unit': unit,
'precision': precision,
'show_args': show_args,
'logger': logger
}
return wrapper
return decorator
# Usage with different configurations
@timing(unit='milliseconds', precision=2, show_args=True)
def fibonacci(n):
"""Calculate Fibonacci number recursively."""
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
@timing(unit='microseconds', precision=0)
def quick_calculation(x, y):
"""Perform a quick mathematical operation."""
return x ** 2 + y ** 2
@timing(unit='seconds', show_args=False)
def slow_operation():
"""Simulate a slow operation."""
time.sleep(0.1)
return "Operation complete"
# Test different timing configurations
result1 = fibonacci(10)
result2 = quick_calculation(5, 7)
result3 = slow_operation()
print(f"Results: {result1}, {result2}, {result3}")
Advanced Parameterized Decorators
Cache Decorator with TTL and Size Limits
import time
import threading
from functools import wraps
from collections import OrderedDict
from typing import Any, Callable, Optional
def cache(ttl=None, max_size=128, typed=False):
"""Advanced caching decorator with TTL and size limits."""
def decorator(func):
# Thread-safe cache storage
cache_data = OrderedDict()
cache_times = {}
lock = threading.RLock()
@wraps(func)
def wrapper(*args, **kwargs):
# Create cache key
if typed:
key = (args, tuple(sorted(kwargs.items())),
tuple(type(arg).__name__ for arg in args))
else:
key = (args, tuple(sorted(kwargs.items())))
with lock:
current_time = time.time()
# Check if key exists and is not expired
if key in cache_data:
if ttl is None or (current_time - cache_times[key]) < ttl:
# Move to end (LRU)
cache_data.move_to_end(key)
return cache_data[key]
else:
# Remove expired entry
del cache_data[key]
del cache_times[key]
# Compute result
result = func(*args, **kwargs)
# Store in cache
cache_data[key] = result
cache_times[key] = current_time
# Enforce size limit (LRU eviction)
if max_size is not None and len(cache_data) > max_size:
oldest_key = next(iter(cache_data))
del cache_data[oldest_key]
del cache_times[oldest_key]
return result
# Cache management methods
def cache_info():
with lock:
current_time = time.time()
valid_entries = 0
expired_entries = 0
for key, timestamp in cache_times.items():
if ttl is None or (current_time - timestamp) < ttl:
valid_entries += 1
else:
expired_entries += 1
return {
'size': len(cache_data),
'valid_entries': valid_entries,
'expired_entries': expired_entries,
'max_size': max_size,
'ttl': ttl,
'typed': typed
}
def cache_clear():
with lock:
cache_data.clear()
cache_times.clear()
def cache_cleanup():
"""Remove expired entries."""
with lock:
current_time = time.time()
expired_keys = [
key for key, timestamp in cache_times.items()
if ttl is not None and (current_time - timestamp) >= ttl
]
for key in expired_keys:
del cache_data[key]
del cache_times[key]
# Attach cache management methods
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
wrapper.cache_cleanup = cache_cleanup
return wrapper
return decorator
# Usage examples
@cache(ttl=5, max_size=10, typed=True)
def expensive_calculation(x, y, operation='add'):
"""Perform expensive mathematical operations."""
print(f"Computing: {x} {operation} {y}")
time.sleep(0.1) # Simulate expensive operation
operations = {
'add': x + y,
'multiply': x * y,
'power': x ** y
}
return operations.get(operation, 0)
@cache(ttl=10, max_size=50)
def fetch_user_data(user_id):
"""Simulate fetching user data from database."""
print(f"Fetching data for user {user_id}")
time.sleep(0.05) # Simulate database query
return {
'id': user_id,
'name': f'User {user_id}',
'created_at': time.time()
}
# Test caching behavior
print("=== Testing expensive calculation ===")
result1 = expensive_calculation(5, 3, 'multiply') # Computed
result2 = expensive_calculation(5, 3, 'multiply') # Cache hit
result3 = expensive_calculation(5.0, 3.0, 'multiply') # Computed (different types)
print(f"Cache info: {expensive_calculation.cache_info()}")
print("\n=== Testing user data cache ===")
user1 = fetch_user_data(123) # Computed
user2 = fetch_user_data(123) # Cache hit
user3 = fetch_user_data(456) # Computed
print(f"Cache info: {fetch_user_data.cache_info()}")
# Wait for TTL expiration
print("\nWaiting for cache expiration...")
time.sleep(6)
fetch_user_data.cache_cleanup()
print(f"After cleanup: {fetch_user_data.cache_info()}")
Authorization Decorator with Role-Based Access
from functools import wraps
from enum import Enum
from typing import List, Union, Callable, Optional
class Role(Enum):
GUEST = "guest"
USER = "user"
MODERATOR = "moderator"
ADMIN = "admin"
class Permission(Enum):
READ = "read"
WRITE = "write"
DELETE = "delete"
MANAGE_USERS = "manage_users"
# Role hierarchy and permissions
ROLE_PERMISSIONS = {
Role.GUEST: [Permission.READ],
Role.USER: [Permission.READ, Permission.WRITE],
Role.MODERATOR: [Permission.READ, Permission.WRITE, Permission.DELETE],
Role.ADMIN: [Permission.READ, Permission.WRITE, Permission.DELETE, Permission.MANAGE_USERS]
}
# Global user context (in real app, this would be from session/JWT)
current_user = {
'id': 123,
'username': 'john_doe',
'role': Role.USER
}
def require_auth(roles: Union[Role, List[Role]] = None,
permissions: Union[Permission, List[Permission]] = None,
allow_owner: bool = False,
owner_field: str = 'user_id'):
"""
Authorization decorator with flexible role and permission checking.
Args:
roles: Required roles (any of the specified roles)
permissions: Required permissions (all specified permissions)
allow_owner: Whether resource owner can access regardless of role
owner_field: Field name to check for ownership
"""
# Normalize inputs to lists
if roles is not None:
roles = roles if isinstance(roles, list) else [roles]
if permissions is not None:
permissions = permissions if isinstance(permissions, list) else [permissions]
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
# Check if user is authenticated
if not current_user:
raise PermissionError("Authentication required")
user_role = current_user.get('role')
user_id = current_user.get('id')
# Check ownership if allowed
if allow_owner and owner_field in kwargs:
if kwargs[owner_field] == user_id:
print(f"✓ Access granted: Owner of resource")
return func(*args, **kwargs)
# Check role requirements
if roles is not None:
if user_role not in roles:
role_names = [role.value for role in roles]
raise PermissionError(f"Access denied. Required roles: {role_names}")
# Check permission requirements
if permissions is not None:
user_permissions = ROLE_PERMISSIONS.get(user_role, [])
missing_permissions = [p for p in permissions if p not in user_permissions]
if missing_permissions:
perm_names = [p.value for p in missing_permissions]
raise PermissionError(f"Access denied. Missing permissions: {perm_names}")
# Log successful authorization
auth_info = []
if roles:
auth_info.append(f"role: {user_role.value}")
if permissions:
perm_names = [p.value for p in permissions]
auth_info.append(f"permissions: {perm_names}")
print(f"✓ Access granted for {current_user['username']} ({', '.join(auth_info)})")
return func(*args, **kwargs)
# Store authorization configuration
wrapper._auth_config = {
'roles': roles,
'permissions': permissions,
'allow_owner': allow_owner,
'owner_field': owner_field
}
return wrapper
return decorator
# Usage examples
@require_auth(roles=Role.ADMIN, permissions=Permission.MANAGE_USERS)
def delete_user(user_id):
"""Delete a user account (admin only)."""
return f"User {user_id} deleted"
@require_auth(permissions=[Permission.WRITE], allow_owner=True, owner_field='user_id')
def update_profile(user_id, name, email):
"""Update user profile (owner or write permission)."""
return f"Profile updated for user {user_id}: {name}, {email}"
@require_auth(roles=[Role.USER, Role.MODERATOR, Role.ADMIN])
def create_post(title, content):
"""Create a new post (authenticated users only)."""
return f"Post created: {title}"
@require_auth(permissions=Permission.READ)
def view_content():
"""View content (requires read permission)."""
return "Content displayed"
# Test authorization
print("=== Testing authorization ===")
try:
# This should work - user has read permission
result = view_content()
print(f"View content: {result}")
except PermissionError as e:
print(f"❌ {e}")
try:
# This should work - user can update own profile
result = update_profile(user_id=123, name="John", email="john@example.com")
print(f"Update profile: {result}")
except PermissionError as e:
print(f"❌ {e}")
try:
# This should fail - user doesn't have admin role
result = delete_user(456)
print(f"Delete user: {result}")
except PermissionError as e:
print(f"❌ {e}")
# Change user role and test again
print("\n=== Testing with admin role ===")
current_user['role'] = Role.ADMIN
try:
result = delete_user(456)
print(f"Delete user: {result}")
except PermissionError as e:
print(f"❌ {e}")
# Check decorator configuration
print(f"Delete user auth config: {delete_user._auth_config}")
Best Practices for Parameterized Decorators
1. Support Both Parameterized and Non-Parameterized Usage
from functools import wraps
def flexible_decorator(func=None, *, param1=None, param2='default'):
"""Decorator that works with or without parameters."""
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
print(f"Executing {f.__name__} with param1={param1}, param2={param2}")
return f(*args, **kwargs)
return wrapper
if func is None:
# Called with parameters: @flexible_decorator(param1='value')
return decorator
else:
# Called without parameters: @flexible_decorator
return decorator(func)
# Both usage patterns work
@flexible_decorator
def function_a():
return "A"
@flexible_decorator(param1='custom', param2='value')
def function_b():
return "B"
print(function_a()) # Uses default parameters
print(function_b()) # Uses custom parameters
2. Validate Decorator Parameters
from functools import wraps
from typing import Union, List
def validated_decorator(timeout: Union[int, float] = 30,
retries: int = 3,
allowed_exceptions: List[type] = None):
"""Decorator with parameter validation."""
# Validate parameters
if not isinstance(timeout, (int, float)) or timeout <= 0:
raise ValueError("timeout must be a positive number")
if not isinstance(retries, int) or retries < 0:
raise ValueError("retries must be a non-negative integer")
if allowed_exceptions is None:
allowed_exceptions = [Exception]
if not all(issubclass(exc, BaseException) for exc in allowed_exceptions):
raise ValueError("allowed_exceptions must be exception classes")
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Implementation using validated parameters
return func(*args, **kwargs)
# Store validated configuration
wrapper._config = {
'timeout': timeout,
'retries': retries,
'allowed_exceptions': allowed_exceptions
}
return wrapper
return decorator
# Valid usage
@validated_decorator(timeout=60, retries=5)
def valid_function():
pass
# This would raise ValueError during decoration
try:
@validated_decorator(timeout=-1) # Invalid timeout
def invalid_function():
pass
except ValueError as e:
print(f"Validation error: {e}")
3. Creating Decorator Factories
from functools import wraps
import logging
def create_decorator_family(base_name='decorator'):
"""Factory that creates related decorators with shared configuration."""
def logging_decorator(level=logging.INFO, include_args=False):
"""Create a logging decorator with specified configuration."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
logger = logging.getLogger(f"{base_name}.{func.__module__}")
if include_args:
logger.log(level, f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
else:
logger.log(level, f"Calling {func.__name__}")
try:
result = func(*args, **kwargs)
logger.log(level, f"{func.__name__} completed successfully")
return result
except Exception as e:
logger.error(f"{func.__name__} failed: {e}")
raise
return wrapper
return decorator
def timing_decorator(unit='seconds', precision=4):
"""Create a timing decorator with specified configuration."""
import time
unit_factors = {'seconds': 1, 'milliseconds': 1000, 'microseconds': 1000000}
factor = unit_factors.get(unit, 1)
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = (time.perf_counter() - start) * factor
print(f"{func.__name__} took {elapsed:.{precision}f} {unit}")
return result
return wrapper
return decorator
return {
'logging': logging_decorator,
'timing': timing_decorator
}
# Create decorator families for different modules
api_decorators = create_decorator_family('api')
database_decorators = create_decorator_family('database')
# Use the generated decorators
@api_decorators['logging'](level=logging.DEBUG, include_args=True)
@api_decorators['timing'](unit='milliseconds', precision=2)
def api_call(endpoint, data):
"""Make an API call."""
import time
time.sleep(0.01) # Simulate API call
return f"Response from {endpoint}"
@database_decorators['logging'](level=logging.INFO)
@database_decorators['timing'](unit='microseconds')
def database_query(query):
"""Execute a database query."""
import time
time.sleep(0.005) # Simulate database query
return f"Results for: {query}"
# Configure logging
logging.basicConfig(level=logging.DEBUG)
# Test the decorators
result1 = api_call('/users', {'name': 'John'})
result2 = database_query('SELECT * FROM users')
Common Patterns and Pitfalls
Handling Mutable Default Arguments
from functools import wraps
# ❌ WRONG: Mutable default argument
def bad_decorator(items=[]): # This is shared across all calls!
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
items.append(func.__name__) # Modifies shared list
return func(*args, **kwargs)
return wrapper
return decorator
# ✅ CORRECT: Use None and create new instances
def good_decorator(items=None):
if items is None:
items = [] # Create new list for each decorator instance
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
items.append(func.__name__)
print(f"Function calls: {items}")
return func(*args, **kwargs)
# Give access to the items list
wrapper._items = items
return wrapper
return decorator
# Test the difference
@good_decorator(['initial'])
def func_a():
pass
@good_decorator(['other'])
def func_b():
pass
func_a() # Function calls: ['initial', 'func_a']
func_b() # Function calls: ['other', 'func_b']
Error Handling in Parameterized Decorators
from functools import wraps
import sys
def robust_decorator(error_handler=None, reraise=True, log_errors=True):
"""Decorator with comprehensive error handling."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
error_info = {
'function': func.__qualname__,
'args': args,
'kwargs': kwargs,
'exception': e,
'exception_type': type(e).__name__
}
if log_errors:
print(f"❌ Error in {func.__qualname__}: {e}", file=sys.stderr)
if error_handler:
try:
result = error_handler(error_info)
if result is not None:
return result
except Exception as handler_error:
print(f"❌ Error handler failed: {handler_error}", file=sys.stderr)
if reraise:
raise
else:
return None
return wrapper
return decorator
# Custom error handlers
def api_error_handler(error_info):
"""Handle API-related errors."""
if isinstance(error_info['exception'], ConnectionError):
return {'error': 'Service temporarily unavailable', 'retry_after': 60}
return None
def calculation_error_handler(error_info):
"""Handle calculation errors."""
if isinstance(error_info['exception'], (ValueError, TypeError)):
return 0 # Return default value
return None
# Usage examples
@robust_decorator(error_handler=api_error_handler, reraise=False)
def make_api_request(url):
"""Make an API request that might fail."""
raise ConnectionError("Network unreachable")
@robust_decorator(error_handler=calculation_error_handler, log_errors=True)
def divide_numbers(a, b):
"""Divide two numbers with error handling."""
return a / b
# Test error handling
result1 = make_api_request("https://api.example.com")
print(f"API result: {result1}")
try:
result2 = divide_numbers(10, 0)
print(f"Division result: {result2}")
except ZeroDivisionError as e:
print(f"Division error caught: {e}")
Conclusion
Creating decorators with parameters and arguments significantly enhances the flexibility and reusability of your Python code. Key takeaways include:
Essential Patterns:
- Use the three-layer structure: parameter function → decorator function → wrapper function
- Always use
@functools.wraps
to preserve function metadata - Validate decorator parameters early to catch configuration errors
- Support both parameterized and non-parameterized usage when possible
Advanced Features:
- Thread-safe caching with TTL and size limits
- Role-based authorization with flexible permission checking
- Comprehensive error handling with custom error handlers
- Decorator factories for creating related decorator families
Best Practices:
- Avoid mutable default arguments; use
None
and create instances as needed - Store configuration in wrapper functions for introspection
- Provide clear error messages for invalid parameters
- Document parameter types and expected behavior
- Consider performance implications of complex decorators
Common Use Cases:
- Caching with configurable expiration and size limits
- Retry logic with customizable attempts and delays
- Authorization and access control
- Logging and monitoring with flexible output options
- Input validation with customizable rules
By mastering parameterized decorators, you'll be able to create powerful, reusable components that can adapt to different requirements while maintaining clean, readable code. These patterns form the foundation for building sophisticated decorator-based frameworks and utilities in Python.
Add Comment
No comments yet. Be the first to comment!