Comprehensive Collection of Python Decorators for Logging, Timing, Error Handling, Caching, and More

The article introduces a wide range of reusable Python decorators that address common cross‑cutting concerns in software development.

1. Logging decorator records function calls and results, handling exceptions and logging stack traces.

import functools
import time
import logging
import traceback
from typing import Callable, Any
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def log_decorator(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        logging.info(f"Calling function {func.__name__} with args: {args}, kwargs: {kwargs}")
        try:
            result = func(*args, **kwargs)
            logging.info(f"Function {func.__name__} returned: {result}")
            return result
        except Exception as e:
            logging.error(f"Function {func.__name__} raised an exception: {e}")
            logging.error(traceback.format_exc())
            raise
    return wrapper

@log_decorator
def add(x: int, y: int) -> int:
    return x + y

2. Performance timing decorator measures execution time and logs it.

def timing_decorator(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        logging.info(f"Function {func.__name__} took {end_time - start_time:.4f} seconds to execute")
        return result
    return wrapper

@timing_decorator
def slow_function():
    time.sleep(2)

3. Exception handling decorator catches exceptions, logs error details, and returns None on failure.

def exception_handler_decorator(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(f"An error occurred in function {func.__name__}: {e}")
            logging.error(traceback.format_exc())
            return None
    return wrapper

@exception_handler_decorator
def risky_function(divisor: int) -> float:
    return 10 / divisor

4. Cache results decorator stores function outputs in a dictionary keyed by arguments.

def cache_results(func: Callable) -> Callable:
    cache = {}
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        key = args + tuple(kwargs.items())
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    return wrapper

@cache_results
def expensive_function(x: int) -> int:
    time.sleep(1)
    return x * x

5. Retry mechanism decorator automatically retries a function upon failure with configurable attempts and delay.

def retry_decorator(max_retries: int = 3, delay: float = 1):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            attempts = 0
            while attempts < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    logging.warning(f"Attempt {attempts}/{max_retries} failed for function {func.__name__}: {e}")
                    time.sleep(delay)
            logging.error(f"All retries failed for function {func.__name__}")
            raise
        return wrapper
    return decorator

@retry_decorator(max_retries=3, delay=1)
def unstable_function():
    if random.choice([True, False]):
        raise Exception("Random failure")
    return "Success"

6. Authentication check decorator validates a token argument before allowing function execution.

def auth_required(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        token = kwargs.get('token')
        if not token or not validate_token(token):
            logging.error("Authentication required")
            raise PermissionError("Authentication required")
        return func(*args, **kwargs)
    return wrapper

def validate_token(token: str) -> bool:
    return token == "valid_token"

@auth_required
def secure_function(data: str, token: str) -> str:
    return f"Processed data: {data}"

7. Parameter validation decorator ensures required keyword arguments are present.

def validate_params(param_names: list):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            for param_name in param_names:
                if param_name not in kwargs:
                    logging.error(f"Missing parameter: {param_name}")
                    raise ValueError(f"Missing parameter: {param_name}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_params(['data'])
def process_data(**kwargs) -> str:
    return f"Processing data: {kwargs['data']}"

8. Decorator chaining combines multiple decorators into a single wrapper.

def combined_decorators(decorator_list: list):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            for dec in reversed(decorator_list):
                func = dec(func)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@combined_decorators([log_decorator, timing_decorator, exception_handler_decorator])
def complex_function(x: int, y: int) -> int:
    return x / y

9. JSON response formatting decorator wraps a function's return value in a standard JSON structure.

def json_response_decorator(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> dict:
        result = func(*args, **kwargs)
        return {"status": "success", "data": result}
    return wrapper

@json_response_decorator
def fetch_data() -> str:
    return "Sample Data"

10. Rate limiting decorator restricts how often a function can be called within a time window.

class RateLimiter:
    def __init__(self, rate_limit: int = 10, period: int = 60):
        self.rate_limit = rate_limit
        self.period = period
        self.requests = []
    def allow_request(self) -> bool:
        current_time = time.time()
        self.requests = [req for req in self.requests if current_time - req < self.period]
        if len(self.requests) < self.rate_limit:
            self.requests.append(current_time)
            return True
        return False

rate_limiter = RateLimiter(rate_limit=5, period=10)

def rate_limit_decorator(limiter: RateLimiter):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            if limiter.allow_request():
                return func(*args, **kwargs)
            else:
                logging.warning("Rate limit exceeded")
                raise RuntimeError("Rate limit exceeded")
        return wrapper
    return decorator

@rate_limit_decorator(rate_limiter)
def limited_function() -> str:
    return "Allowed Request"

11. Environment variable injection decorator adds specified environment variables to function keyword arguments.

import os

def inject_env_vars(env_var_names: list):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            env_vars = {var: os.getenv(var) for var in env_var_names}
            kwargs.update(env_vars)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@inject_env_vars(['API_KEY'])
def use_api_key(api_key: str) -> str:
    return f"Using API Key: {api_key}"

12. Response time monitoring decorator logs a warning when execution exceeds a threshold.

def response_time_monitoring_decorator(threshold: float):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            response_time = end_time - start_time
            if response_time > threshold:
                logging.warning(f"Response time for function {func.__name__} exceeded threshold: {response_time}s")
            return result
        return wrapper
    return decorator

@response_time_monitoring_decorator(threshold=1)
def slow_response() -> str:
    time.sleep(2)
    return "Slow Response"

13. Custom header addition decorator merges user‑provided headers with predefined ones.

def add_custom_headers(headers: dict):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            existing_headers = kwargs.get('headers', {})
            updated_headers = {**existing_headers, **headers}
            kwargs['headers'] = updated_headers
            return func(*args, **kwargs)
        return wrapper
    return decorator

@add_custom_headers({'Authorization': 'Bearer token'})
def make_request(headers: dict) -> dict:
    return headers

14. Data transformation decorator applies a user‑defined transformation to both positional and keyword arguments before invoking the function.

def data_transformation_decorator(transform_func: Callable):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            transformed_args = transform_func(args)
            transformed_kwargs = transform_func(kwargs)
            return func(*transformed_args, **transformed_kwargs)
        return wrapper
    return decorator

def transform_example(data):
    return {k: v.upper() if isinstance(v, str) else v for k, v in data.items()}

@data_transformation_decorator(transform_example)
def print_transformed_data(data: dict) -> dict:
    return data

15. Concurrency control decorator limits the number of simultaneous executions using a semaphore.

import threading

def concurrency_control_decorator(max_concurrent: int):
    semaphore = threading.Semaphore(max_concurrent)
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            with semaphore:
                return func(*args, **kwargs)
        return wrapper
    return decorator

@concurrency_control_decorator(max_concurrent=3)
def concurrent_task(task_id: int) -> str:
    time.sleep(1)
    return f"Task {task_id} completed"

16. Distributed lock decorator uses Redis to ensure only one process executes a critical section at a time.

from redis import Redis
redis_client = Redis()

def distributed_lock_decorator(lock_key: str, timeout: int = 10):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            lock_acquired = redis_client.setnx(lock_key, 1)
            if lock_acquired:
                redis_client.expire(lock_key, timeout)
                try:
                    return func(*args, **kwargs)
                finally:
                    redis_client.delete(lock_key)
            else:
                logging.warning(f"Distributed lock for {lock_key} already acquired")
                raise RuntimeError("Distributed lock already acquired")
        return wrapper
    return decorator

@distributed_lock_decorator(lock_key='my_lock')
def critical_section() -> str:
    time.sleep(2)
    return "Critical Section Completed"

17. API version control decorator validates that the caller supplies the required API version.

def api_version_control_decorator(required_version: str):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            version = kwargs.get('version')
            if version != required_version:
                logging.error(f"Incorrect API version: {version}. Required version: {required_version}")
                raise ValueError(f"Incorrect API version: {version}. Required version: {required_version}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@api_version_control_decorator(required_version='v1')
def api_call(version: str) -> str:
    return "API Call Successful"

18. Security audit decorator logs function entry, arguments, and result for audit purposes.

def security_audit_decorator(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        audit_log = f"Audit Log - Function: {func.__name__}, Args: {args}, Kwargs: {kwargs}"
        logging.info(audit_log)
        result = func(*args, **kwargs)
        logging.info(f"Audit Log - Result: {result}")
        return result
    return wrapper

@security_audit_decorator
def sensitive_operation(data: str) -> str:
    return f"Processed Sensitive Data: {data}"

19. Input validation decorator validates incoming data against a schema before calling the function.

def input_validation_decorator(schema: dict):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            validated_data = validate_input(schema, kwargs)
            return func(*args, **validated_data)
        return wrapper
    return decorator

def validate_input(schema: dict, data: dict) -> dict:
    for key, expected_type in schema.items():
        if key not in data or not isinstance(data[key], expected_type):
            raise ValueError(f"Invalid input for {key}. Expected type: {expected_type}")
    return data

@input_validation_decorator({'name': str, 'age': int})
def create_user(name: str, age: int) -> str:
    return f"User created: {name}, Age: {age}"

20. Output filtering decorator applies a filter function to the result before returning it.

def output_filtering_decorator(filter_func: Callable):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            result = func(*args, **kwargs)
            filtered_result = filter_func(result)
            return filtered_result
        return wrapper
    return decorator

def filter_example(data):
    return {k: v for k, v in data.items() if v is not None}

@output_filtering_decorator(filter_example)
def get_filtered_data() -> dict:
    return {"name": "John", "age": None, "city": "New York"}

The article concludes with a test harness that demonstrates each decorator in action, showing typical usage patterns and expected console output.