Decorators and Generators in Python: The Advanced Guide

Decorators and Generators are advanced Python features that separate intermediate programmers from true professionals. Decorators allow you to elegantly modify the behavior of functions, while Generators create memory efficient iterators. In this comprehensive guide, you will master both of these concepts with practical, real world examples.

🎨 What Are Decorators?

Decorators are essentially functions that modify the behavior of other functions without actually altering their source code. Think of them as wrappers that add extra functionality to an existing function.

# A simple decorator function
def my_decorator(func):
    def wrapper():
        print("🚀 Before the function execution")
        func()
        print("✅ After the function execution")
    return wrapper

@my_decorator
def greeting():
    print("Hello, Python Universe!")

# Calling the decorated function
greeting()
# Output:
# 🚀 Before the function execution
# Hello, Python Universe!
# ✅ After the function execution

The @ symbol is simply "syntactic sugar" for the following operation:

# This syntax:
@my_decorator
def greeting():
    print("Hello!")

# Is exactly equivalent to:
def greeting():
    print("Hello!")
greeting = my_decorator(greeting)

📝 Creating Practical Decorators

Decorator with Function Arguments

If the function you are decorating accepts arguments, your wrapper must accept them as well. We use *args and **kwargs for this.

def log_call(func):
    """Logs exactly when a function is called and its arguments"""
    def wrapper(*args, **kwargs):
        print(f"📞 Calling function: {func.__name__}")
        print(f"   Positional Args: {args}")
        print(f"   Keyword Args: {kwargs}")

        result = func(*args, **kwargs)

        print(f"   Returned: {result}")
        return result
    return wrapper

@log_call
def add_numbers(a, b):
    return a + b

@log_call
def custom_greeting(name, greeting_text="Hello"):
    return f"{greeting_text}, {name}!"

print(add_numbers(10, 5))
print(custom_greeting("Python", greeting_text="Welcome"))

Preserving Metadata with functools.wraps

When you decorate a function, it loses its original metadata (like its name and docstring). To fix this, always use @wraps from the built-in functools library.

from functools import wraps

def professional_decorator(func):
    @wraps(func)  # Preserves __name__, __doc__, etc.
    def wrapper(*args, **kwargs):
        """Internal wrapper documentation"""
        return func(*args, **kwargs)
    return wrapper

@professional_decorator
def my_function():
    """Original function documentation"""
    pass

print(my_function.__name__)  # Prints 'my_function' (not 'wrapper')
print(my_function.__doc__)   # Prints 'Original function documentation'

Decorators that Accept Parameters

from functools import wraps

def repeat_execution(times):
    """Decorator that repeats the function N times"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(times):
                result = func(*args, **kwargs)
                results.append(result)
            return results
        return wrapper
    return decorator

@repeat_execution(times=3)
def say_hi():
    print("Hi!")
    return "said hi"

results = say_hi()
# Output: Hi! (printed 3 times)
print(results)  # ['said hi', 'said hi', 'said hi']

⚡ Highly Useful Practical Decorators

Timer: Measuring Execution Time

import time
from functools import wraps

def timer(func):
    """Measures the exact execution time of a function"""
    @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__} executed in {duration:.4f} seconds")
        return result
    return wrapper

@timer
def process_data():
    time.sleep(1)  # Simulating heavy data processing
    return "Data processed successfully"

process_data()  # ⏱️ process_data executed in 1.0012 seconds

Cache and Memoization

If you have an expensive function, you can cache its results to avoid repeating calculations.

from functools import wraps

def simple_cache(func):
    """A simple dictionary based cache for function results"""
    memory = {}

    @wraps(func)
    def wrapper(*args):
        if args in memory:
            print(f"📦 Cache hit for {args}")
            return memory[args]

        result = func(*args)
        memory[args] = result
        print(f"💾 Storing new result in cache: {args}")
        return result
    return wrapper

@simple_cache
def fibonacci_recursive(n):
    """Calculates Fibonacci sequence recursively"""
    if n < 2:
        return n
    return fibonacci_recursive(n-1) + fibonacci_recursive(n-2)

print(fibonacci_recursive(10))  # Extremely fast thanks to caching!

# Note: Python 3.9+ includes a built in lru_cache which is heavily optimized:
from functools import lru_cache

@lru_cache(maxsize=128)
def optimized_fibonacci(n):
    if n < 2:
        return n
    return optimized_fibonacci(n-1) + optimized_fibonacci(n-2)

Retry Mechanism

import time
from functools import wraps

def retry_on_failure(attempts=3, delay=1, exceptions=(Exception,)):
    """Retries executing a function multiple times if it fails"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(1, attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    print(f"⚠️ Attempt {attempt}/{attempts} failed: {e}")

                    if attempt < attempts:
                        print(f"   Waiting {delay} seconds before retrying...")
                        time.sleep(delay)

            raise last_exception
        return wrapper
    return decorator

@retry_on_failure(attempts=3, delay=1)
def fetch_api_data():
    import random
    if random.random() < 0.7:  # 70% chance of failure
        raise ConnectionError("Network connection failed")
    return {"status": "success"}

try:
    result = fetch_api_data()
    print(f"✅ Success: {result}")
except ConnectionError as e:
    print(f"❌ Failed permanently after all attempts: {e}")

Authentication and Authorization

from functools import wraps

# Simulating a logged in user session
current_user = {"name": "Admin", "role": "admin"}

def requires_login(func):
    """Verifies if the user is currently logged in"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        if not current_user:
            raise PermissionError("❌ Please login first!")
        return func(*args, **kwargs)
    return wrapper

def requires_role(*allowed_roles):
    """Verifies if the logged in user has the required permission"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if current_user.get("role") not in allowed_roles:
                raise PermissionError(
                    f"❌ Access denied. Required roles: {allowed_roles}"
                )
            return func(*args, **kwargs)
        return wrapper
    return decorator

@requires_login
@requires_role("admin", "moderator")
def delete_user_account(user_id):
    return f"✅ User account {user_id} successfully deleted"

print(delete_user_account(42))

🔄 What Are Generators?

Generators are special functions that produce a sequence of values on demand, utilizing the yield keyword instead of return. They are incredibly memory efficient because they only generate one single value at a time, rather than storing everything in memory simultaneously.

# Standard function - stores everything in memory
def create_list(n):
    result = []
    for i in range(n):
        result.append(i ** 2)
    return result

# Generator function - generates items on demand
def create_generator(n):
    for i in range(n):
        yield i ** 2

# Memory comparison
import sys

standard_list = create_list(1000000)
gen_object = create_generator(1000000)

print(f"List Memory: {sys.getsizeof(standard_list):,} bytes")  # ~8MB
print(f"Generator Memory: {sys.getsizeof(gen_object)} bytes")  # ~120 bytes!

📝 Building Generators

Simple Generator Example

def countdown_timer(n):
    """A countdown generator function"""
    print("🚀 Initiating countdown sequence...")
    while n > 0:
        yield n
        n -= 1
    print("🔥 Liftoff!")

# Using it in a for loop
for number in countdown_timer(5):
    print(number)

# Or advancing manually using next()
gen = countdown_timer(3)
print(next(gen))  # 3
print(next(gen))  # 2
print(next(gen))  # 1
# print(next(gen))  # Raises StopIteration exception

Infinite Generators

def infinite_even_numbers():
    """An infinite generator producing even numbers"""
    n = 0
    while True:
        yield n
        n += 2

# Use with caution and break conditions!
evens = infinite_even_numbers()
for _ in range(10):
    print(next(evens), end=" ")
# Output: 0 2 4 6 8 10 12 14 16 18

Generators with Multiple Yields

def read_large_file_in_chunks(filepath, chunk_size=1024):
    """Reads a massive file efficiently in chunks"""
    with open(filepath, 'r') as file:
        while True:
            chunk = file.read(chunk_size)
            if not chunk:
                break
            yield chunk

# You can process a 10GB file without crashing your memory
# for chunk in read_large_file_in_chunks("massive_log_file.txt"):
#     process_data(chunk)

This is extremely useful when dealing with data science. Learn more about file manipulation in our guide on handling Python files.

🎯 Generator Expressions

Generator expressions are highly similar to Python list comprehensions, but they use parentheses instead of square brackets.

# List comprehension - creates the full list in memory
squared_list = [x**2 for x in range(1000000)]

# Generator expression - generates lazily on demand
squared_gen = (x**2 for x in range(1000000))

# Can be used directly within functions effortlessly
total_sum = sum(x**2 for x in range(1000))  # No extra brackets needed
average = sum(x for x in range(100)) / 100

⚡ Yield From Delegation

def nested_generator():
    yield from range(3)      # Yields 0, 1, 2
    yield from "abc"         # Yields a, b, c
    yield from [10, 20, 30]  # Yields 10, 20, 30

for item in nested_generator():
    print(item, end=" ")
# Output: 0 1 2 a b c 10 20 30

# A brilliant practical example: flattening deeply nested lists
def flatten_list(nested_list):
    """Recursively flattens nested lists"""
    for item in nested_list:
        if isinstance(item, list):
            yield from flatten_list(item)
        else:
            yield item

complex_list = [1, [2, 3, [4, 5]], 6, [7, 8]]
print(list(flatten_list(complex_list)))
# Output: [1, 2, 3, 4, 5, 6, 7, 8]

🎯 Practical Project: Advanced Data Processing Pipeline

Let us construct a sophisticated data processing system using decorators and generators, applying modern Object Oriented Programming (OOP) concepts.

import time
import random
from functools import wraps
from typing import Generator, Callable, List
from dataclasses import dataclass

# ============================================
# SYSTEM DECORATORS
# ============================================

def pipeline_timer(func):
    """Measures pipeline execution time"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        duration = time.perf_counter() - start
        print(f"⏱️ {func.__name__} finished in: {duration:.4f}s")
        return result
    return wrapper

def log_pipeline_step(func):
    """Logs pipeline execution steps"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"📍 Starting Step: {func.__name__}")
        result = func(*args, **kwargs)
        print(f"✅ Completed Step: {func.__name__}")
        return result
    return wrapper

# ============================================
# PIPELINE GENERATORS
# ============================================

@dataclass
class RawData:
    id: int
    value: float
    timestamp: float

def simulate_data_stream(amount: int) -> Generator[RawData, None, None]:
    """Generator that simulates an active data source"""
    for i in range(amount):
        yield RawData(
            id=i,
            value=random.uniform(0, 100),
            timestamp=time.time()
        )
        time.sleep(0.01)  # Simulating network latency

def filter_low_values(data_stream: Generator, minimum_threshold: float = 20) -> Generator:
    """Filters out data below the minimum threshold"""
    for data in data_stream:
        if data.value >= minimum_threshold:
            yield data

def transform_to_dict(data_stream: Generator) -> Generator[dict, None, None]:
    """Transforms raw object data into a processed dictionary format"""
    for data in data_stream:
        yield {
            "id": data.id,
            "original_value": data.value,
            "processed_value": round(data.value * 1.1, 2),
            "category": "High" if data.value > 50 else "Low",
            "processed_at": time.time()
        }

def batch_data(data_stream: Generator, batch_size: int = 10) -> Generator[List, None, None]:
    """Groups processed data into specific batch sizes"""
    batch = []
    for data in data_stream:
        batch.append(data)
        if len(batch) >= batch_size:
            yield batch
            batch = []

    if batch:  # Yielding the final partial batch
        yield batch

# ============================================
# MAIN PIPELINE CLASS
# ============================================

class DataPipeline:
    """Core data processing pipeline system"""

    def __init__(self, name: str):
        self.name = name
        self.steps: List[Callable] = []
        self.statistics = {
            "total_items_processed": 0,
            "total_batches_generated": 0
        }

    def add_step(self, step_function: Callable) -> 'DataPipeline':
        """Adds a processing step using a fluent interface"""
        self.steps.append(step_function)
        return self

    @pipeline_timer
    @log_pipeline_step
    def execute_pipeline(self, input_stream: Generator) -> Generator:
        """Executes the entire pipeline sequentially"""
        stream = input_stream

        for step in self.steps:
            stream = step(stream)

        return stream

    def gather_results(self, processed_stream: Generator) -> List[dict]:
        """Consumes the generator and gathers the final results"""
        final_results = []

        for batch in processed_stream:
            self.statistics["total_batches_generated"] += 1

            for item in batch:
                final_results.append(item)
                self.statistics["total_items_processed"] += 1

        return final_results

    def display_report(self):
        """Displays the pipeline execution statistics"""
        print(f"\n📊 REPORT: {self.name}")
        print("=" * 40)
        for key, value in self.statistics.items():
            print(f"  {key}: {value}")

# ============================================
# EXECUTION DEMONSTRATION
# ============================================

if __name__ == "__main__":
    print("=" * 60)
    print("🚀 ADVANCED DATA PIPELINE SYSTEM")
    print("=" * 60)

    # Initialize the pipeline
    pipeline = DataPipeline("Enterprise Sales Pipeline")

    # Configure the processing steps
    pipeline.add_step(
        lambda data: filter_low_values(data, minimum_threshold=25)
    ).add_step(
        transform_to_dict
    ).add_step(
        lambda data: batch_data(data, batch_size=5)
    )

    # Generate the incoming data stream
    incoming_stream = simulate_data_stream(50)

    # Execute the pipeline (generators make this lazy and instantaneous)
    print("\n" + "-" * 60)
    processed_stream = pipeline.execute_pipeline(incoming_stream)

    # Gather the results (this is where actual computation happens)
    final_results = pipeline.gather_results(processed_stream)

    # Display the final report
    pipeline.display_report()

    print("\n📄 First 3 Processed Results:")
    for item in final_results[:3]:
        print(f"  {item}")

    print(f"\n✅ Total valid items processed: {len(final_results)}")

This project is perfect for adding to your Python portfolio, as it demonstrates a deep understanding of memory management and clean architecture.

💡 When to Use Which Tool

You should use Decorators specifically to:

• ✅ Implement logging and monitoring systems automatically.
• ✅ Create caching and memoization layers.
• ✅ Handle authentication and role based authorization checks.
• ✅ Perform strict input validation before function execution.
• ✅ Add network retry logic without cluttering core business logic.
• ✅ Measure execution time precisely.

You should use Generators specifically to:

• ✅ Process extremely large files that exceed available RAM.
• ✅ Handle live streaming data continuously.
• ✅ Build complex, memory efficient data processing pipelines.
• ✅ Represent infinite mathematical sequences.
• ✅ Save significant amounts of system memory gracefully.

For official technical documentation regarding these features, always check the Python functools documentation.

Conclusion

Mastering Decorators and Generators will transform your Python code from functional to highly professional. They provide elegant solutions to complex architectural problems while ensuring your applications remain highly performant and easy to maintain.