Amazing Decorator Functions

Core Purpose: Elegant "Wrapping"

A decorator is a core feature of Python that is essentially a callable object (usually a function) that takes a function as an argument and returns a new function. Its core purpose is: to dynamically add additional functionality to the decorated function (or class) without modifying the original code. This perfectly follows the "Open-Closed Principle" in software development:

• Open for extension: you can easily add new behaviors to existing functionality.
• Closed for modification: no need to change the original, already working code.

You can think of it as gift wrapping. The gift itself (core functionality) hasn't changed, but the wrapping paper (decorator) gives it a more beautiful appearance or additional meaning (extra functionality).

Common Use Cases

Decorators have extremely wide applications. Here are some of the most common and practical scenarios:

Logging

• Purpose: Automatically record function execution information, such as function name, passed parameters, return values, execution time, etc. This is very useful for debugging and monitoring program running status.
• Scenario: You have a function that handles user requests, and you want to know who, when, called this function and what the processing result was each time, but you don't want to write print or logging.info at the beginning and end of every function.

import functools
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_function_call(func):
    """Decorator for logging function call information"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"Calling function: {func.__name__}, args: args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        logger.info(f"Function {func.__name__} completed, return value: {result}")
        return result
    return wrapper

# Usage examples
@log_function_call
def add_numbers(a, b):
    """Add two numbers"""
    return a + b

@log_function_call
def greet(name, greeting="Hello"):
    """Greeting function"""
    return f"{greeting}, {name}!"

# Testing
if __name__ == "__main__":
    print(add_numbers(5, 3))
    print(greet("Alice", greeting="Hi"))

Performance Benchmarking / Timing

• Purpose: Measure and output function execution time to help you find performance bottlenecks in your code.
• Scenario: You want to optimize a piece of code but aren't sure which function is running slowly. You can use a @timer decorator to quickly decorate several suspicious functions to precisely measure their execution time.

import functools
import time

def timer(func):
    """Decorator for measuring function execution time"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        end_time = time.perf_counter()
        execution_time = end_time - start_time
        print(f"Function {func.__name__} execution time: {execution_time:.4f} seconds")
        return result
    return wrapper

# Usage examples
@timer
def slow_function():
    """Simulate a time-consuming operation"""
    time.sleep(2)
    return "Task completed"

@timer
def fast_function():
    """Fast execution function"""
    return "Quick completion"

# Testing
if __name__ == "__main__":
    print(slow_function())
    print(fast_function())

Timing

• Purpose: Precisely measure the execution time of functions or code blocks, providing simple and intuitive performance analysis information.
• Scenario: When optimizing algorithm performance, you need to compare the time differences between different implementation solutions. Or during development, you want to quickly understand function execution efficiency to help identify performance bottlenecks.

import functools
import time
from contextlib import contextmanager

def timing(name=None):
    """Precise timing decorator"""
    def decorator(func):
        func_name = name or func.__name__
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.perf_counter()
            result = func(*args, **kwargs)
            end_time = time.perf_counter()
            duration = end_time - start_time
            print(f"⏱️ {func_name} execution time: {duration:.4f}s")
            return result
        
        return wrapper
    return decorator

@contextmanager
def time_block(name):
    """Code block timing context manager"""
    start_time = time.perf_counter()
    try:
        yield
    finally:
        end_time = time.perf_counter()
        duration = end_time - start_time
        print(f"⏱️ {name} code block time: {duration:.4f}s")

# Usage examples
@timing("Fibonacci calculation")
def fibonacci_recursive(n):
    """Recursive Fibonacci calculation"""
    if n <= 1:
        return n
    return fibonacci_recursive(n-1) + fibonacci_recursive(n-2)

@timing("Fast calculation")
def fibonacci_iterative(n):
    """Iterative Fibonacci calculation"""
    if n <= 1:
        return n
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b

@timing()
def process_data():
    """Simulate data processing"""
    time.sleep(0.1)  # Simulate processing delay
    return "Processing completed"

# Testing
if __name__ == "__main__":
    print("=== Timing Decorator Test ===")
    
    # Compare different algorithm performance
    print("\nAlgorithm performance comparison:")
    result1 = fibonacci_recursive(8)  # Smaller value to avoid long execution time
    result2 = fibonacci_iterative(30)
    print(f"Results: {result1}, {result2}")
    
    # Function execution timing
    print("\nFunction execution timing:")
    process_data()
    
    # Code block timing
    print("\nCode block timing:")
    with time_block("Batch calculation"):
        total = 0
        for i in range(1000000):
            total += i
        print(f"Calculation result: {total}")

Authentication & Authorization

• Purpose: Check before function execution whether the user is logged in or has permission to perform the operation. If validation fails, prevent function execution and redirect to login page or return error information.
• Scenario: Very common in web development (such as Django, Flask frameworks). For example, a @login_required decorator can ensure that only logged-in users can access certain view functions (pages).

import functools

# Mock user database
users = {
    "admin": {"password": "admin123", "role": "admin"},
    "user1": {"password": "pass123", "role": "user"},
    "guest": {"password": "guest", "role": "guest"}
}

# Current logged-in user (mock session)
current_user = None

def login_required(required_role="user"):
    """Decorator requiring user login"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if current_user is None:
                raise PermissionError("Please log in first!")
            
            user_role = users.get(current_user, {}).get("role")
            if user_role is None:
                raise PermissionError("User does not exist or role not defined")
            
            # Fixed permission check logic - using role hierarchy system
            role_hierarchy = {"guest": 0, "user": 1, "admin": 2}
            required_level = role_hierarchy.get(required_role, 1)
            user_level = role_hierarchy.get(user_role, 0)
            
            if user_level < required_level:
                raise PermissionError(f"Requires {required_role} permission, current user permission: {user_role}")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage examples
@login_required()
def view_profile():
    """View user profile"""
    return f"Welcome to view your profile, {current_user}!"

@login_required("admin")
def admin_panel():
    """Admin panel"""
    return "Welcome to the admin panel!"

# Testing
if __name__ == "__main__":
    # Mock login
    current_user = "user1"
    print(view_profile())
    
    try:
        print(admin_panel())
    except PermissionError as e:
        print(f"Permission error: {e}")
    
    # Test logout situation
    current_user = None
    try:
        print(view_profile())
    except PermissionError as e:
        print(f"Permission error: {e}")

Input Validation & Sanitization

• Purpose: Before function execution, check whether the passed parameters are valid (such as whether the type is correct, whether numerical values are within valid range), and clean or convert parameters.
• Scenario: A function that calculates circle area needs to ensure the radius is a non-negative number. You can use a decorator to validate parameters, automatically throwing an exception or returning an error if the radius is negative, letting the core function focus on pure calculation logic.

import functools

def validate_input(**validations):
    """Decorator for validating function parameters"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Merge args and kwargs into dictionary for validation
            all_args = dict(zip(func.__code__.co_varnames, args))
            all_args.update(kwargs)
            
            # Validate each parameter
            for param_name, validation in validations.items():
                if param_name in all_args:
                    value = all_args[param_name]
                    if not validation(value):
                        raise ValueError(f"Parameter {param_name} with value {value} does not meet validation rules")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Validation functions
def is_positive(number):
    return number > 0

def is_non_empty_string(text):
    return isinstance(text, str) and len(text.strip()) > 0

def is_valid_age(age):
    return isinstance(age, int) and 0 < age < 150

# Usage examples
@validate_input(radius=is_positive, height=is_positive)
def calculate_cylinder_volume(radius, height):
    """Calculate cylinder volume"""
    return 3.14159 * radius ** 2 * height

@validate_input(name=is_non_empty_string, age=is_valid_age)
def create_user_profile(name, age):
    """Create user profile"""
    return f"User: {name}, Age: {age}"

# Testing
if __name__ == "__main__":
    try:
        print(calculate_cylinder_volume(5, 10))
        print(create_user_profile("Alice", 25))
        
        # Test invalid input
        print(calculate_cylinder_volume(-5, 10))  # Will throw exception
    except ValueError as e:
        print(f"Input validation error: {e}")

Retry Mechanism

• Purpose: When function execution fails (such as network fluctuations, temporary database connection issues), automatically retry a specified number of times.
• Scenario: Calling a third-party API interface, but this interface is occasionally unstable. You can use a @retry(times=3) decorator to let it automatically retry 3 times after failure instead of immediately reporting an error.

import functools
import time
import random

def retry(max_attempts=3, delay=1, backoff=2, exceptions=(Exception,)):
    """Decorator for automatic retry on function execution failure"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            current_delay = delay
            
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    attempts += 1
                    if attempts == max_attempts:
                        print(f"Failed after {max_attempts} retry attempts: {e}")
                        raise
                    
                    print(f"Attempt {attempts} failed: {e}, retrying after {current_delay} seconds...")
                    time.sleep(current_delay)
                    current_delay *= backoff  # Exponential backoff
        return wrapper
    return decorator

# Usage examples
@retry(max_attempts=3, delay=1, exceptions=(ValueError,))
def unreliable_api_call():
    """Simulate unreliable API call"""
    if random.random() < 0.7:  # 70% probability of failure
        raise ValueError("API call failed: connection timeout")
    return "API call successful!"

@retry(max_attempts=5, delay=2, exceptions=(RuntimeError,))
def process_data():
    """Simulate data processing"""
    if random.random() < 0.5:
        raise RuntimeError("Data processing error")
    return "Data processing completed"

# Testing
if __name__ == "__main__":
    try:
        print(unreliable_api_call())
    except Exception as e:
        print(f"Final failure: {e}")
    
    try:
        print(process_data())
    except Exception as e:
        print(f"Final failure: {e}")

Function Registration

• Purpose: Automatically register functions to a central repository (such as plugin systems, routing systems) without needing to explicitly call registration code.
• Scenario: In web frameworks, use @app.route('/url') decorator to register a function as a controller for handling specific URL requests. The framework will automatically collect all functions decorated with this decorator when starting up to build a routing mapping table.

import functools

class PluginRegistry:
    """Plugin registry"""
    def __init__(self):
        self.plugins = {}
    
    def register(self, name=None):
        """Decorator for registering plugins"""
        def decorator(func):
            plugin_name = name or func.__name__
            self.plugins[plugin_name] = func
            return func
        return decorator
    
    def execute_plugin(self, name, *args, **kwargs):
        """Execute specified plugin"""
        if name not in self.plugins:
            raise ValueError(f"Plugin {name} not registered")
        return self.plugins[name](*args, **kwargs)
    
    def list_plugins(self):
        """List all registered plugins"""
        return list(self.plugins.keys())

# Create registry instance
registry = PluginRegistry()

# Use decorator to register plugins
@registry.register("text_processor")
def process_text(text):
    """Text processing plugin"""
    return text.upper()

@registry.register("math_operation")
def calculate_square(x):
    """Square calculation plugin"""
    return x ** 2

@registry.register()  # Use function name as plugin name
def custom_format(data):
    """Custom formatting plugin"""
    return f"Formatted result: {data}"

# Testing
if __name__ == "__main__":
    print("Registered plugins:", registry.list_plugins())
    
    # Execute plugins
    print(registry.execute_plugin("text_processor", "hello world"))
    print(registry.execute_plugin("math_operation", 5))
    print(registry.execute_plugin("custom_format", "test data"))

Caching (Memoization)

• Purpose: Cache function results. When called again with the same parameters, directly return cached results without recalculating, significantly improving performance of computation-intensive functions.
• Scenario: Calculating Fibonacci sequence. Using @functools.lru_cache, this built-in decorator can avoid a lot of repeated calculations.

import functools
import time

def cache_results(max_size=128):
    """Decorator for caching function results"""
    def decorator(func):
        cache = {}
        access_order = []  # Record access order for LRU strategy
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Create cache key
            cache_key = (args, frozenset(kwargs.items()) if kwargs else frozenset())
            
            if cache_key in cache:
                print(f"Cache hit: {func.__name__}{args}")
                # Update access order
                access_order.remove(cache_key)
                access_order.append(cache_key)
                return cache[cache_key]
            
            # If cache is full, delete least recently used entry
            if len(cache) >= max_size:
                oldest_key = access_order.pop(0)
                del cache[oldest_key]
            
            result = func(*args, **kwargs)
            cache[cache_key] = result
            access_order.append(cache_key)
            print(f"Result cached: {func.__name__}{args}")
            return result
        
        # Provide cache clearing method
        def clear_cache():
            cache.clear()
            access_order.clear()
        
        wrapper.clear_cache = clear_cache
        wrapper.cache_info = lambda: {"cache_size": len(cache), "max_size": max_size}
        return wrapper
    return decorator

# Usage examples
@cache_results(max_size=10)
def fibonacci(n):
    """Calculate Fibonacci sequence"""
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

@cache_results()
def expensive_calculation(x, y):
    """Simulate expensive calculation"""
    time.sleep(1)  # Simulate calculation time
    return x * y + x + y

# Testing
if __name__ == "__main__":
    # Test Fibonacci (will show cache effects)
    print("fibonacci(10) =", fibonacci(10))
    print("fibonacci(10) =", fibonacci(10))  # This time will get from cache
    
    # Test expensive calculation
    start = time.time()
    result1 = expensive_calculation(5, 3)
    end = time.time()
    print(f"First calculation time: {end-start:.2f}s")
    
    start = time.time()
    result2 = expensive_calculation(5, 3)  # This time will be very fast
    end = time.time()
    print(f"Second calculation time: {end-start:.4f}s")
    
    print(f"Results: {result1}, {result2}")

Transaction Management

• Purpose: Start a database transaction before function execution, commit the transaction if function executes successfully, rollback the transaction if an exception occurs.
• Scenario: Ensure atomicity of database operations. For example, a function needs to update two tables - either both succeed or both fail. Using decorators can manage this boundary well.

import functools

class DatabaseTransaction:
    """Mock database transaction"""
    def __init__(self):
        self.in_transaction = False
        self.changes = []
    
    def begin(self):
        """Start transaction"""
        if self.in_transaction:
            raise RuntimeError("Transaction already in progress")
        self.in_transaction = True
        self.changes = []
        print("Transaction started")
    
    def commit(self):
        """Commit transaction"""
        if not self.in_transaction:
            raise RuntimeError("No transaction in progress")
        print(f"Transaction committed: applied {len(self.changes)} changes")
        self.in_transaction = False
        self.changes = []
    
    def rollback(self):
        """Rollback transaction"""
        if not self.in_transaction:
            raise RuntimeError("No transaction in progress")
        print(f"Transaction rolled back: reverted {len(self.changes)} changes")
        self.in_transaction = False
        self.changes = []
    
    def execute(self, query, params=None):
        """Execute SQL statement"""
        if not self.in_transaction:
            raise RuntimeError("Must execute within transaction")
        
        change = f"Execute: {query} {params or ''}"
        self.changes.append(change)
        print(change)
        return True

# Create database instance
db = DatabaseTransaction()

def with_transaction(func):
    """Decorator for managing database transactions"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            db.begin()
            result = func(*args, **kwargs)
            db.commit()
            return result
        except Exception as e:
            db.rollback()
            print(f"Transaction rollback reason: {e}")
            raise
    return wrapper

# Usage examples
@with_transaction
def transfer_money(from_account, to_account, amount):
    """Transfer operation"""
    # Mock database operations
    db.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_account))
    db.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_account))
    
    # Mock business logic check
    if amount > 1000:
        raise ValueError("Single transfer amount cannot exceed 1000")
    
    db.execute("INSERT INTO transactions VALUES (?, ?, ?)", (from_account, to_account, amount))
    return "Transfer successful"

@with_transaction
def create_user_account(username, email):
    """Create user account"""
    db.execute("INSERT INTO users (username, email) VALUES (?, ?)", (username, email))
    db.execute("INSERT INTO accounts (user_id, balance) VALUES (1, 100)")  # Simplified ID handling
    return f"User {username} created successfully"

# Testing
if __name__ == "__main__":
    try:
        print(transfer_money(1, 2, 500))
        print("---")
        print(transfer_money(3, 4, 1500))  # This will fail and rollback
    except Exception as e:
        print(f"Operation failed: {e}")
    
    print("---")
    try:
        print(create_user_account("alice", "alice@example.com"))
    except Exception as e:
        print(f"Operation failed: {e}")

Summary

Decorators provide a very clear, non-intrusive way to separate core business logic (the function itself) from cross-cutting concerns (auxiliary functionality such as logging, caching, authentication), making code more modular, readable, and maintainable. These examples cover various practical application scenarios of decorators, each of which can be run directly to see the effects. You can modify and extend these decorators according to your needs.