Navigation

Python

Python Decorators with Parameters and Arguments: Advanced Decorator Patterns

Master Python decorators with parameters. Learn to create flexible, configurable decorators with retry logic, caching, timing, and authorization patterns.

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

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:

  1. Outer function: Accepts the decorator parameters
  2. Middle function: The actual decorator that receives the function
  3. 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.

Share this article

Add Comment

No comments yet. Be the first to comment!

More from Python