Python Decorators and Generators — Advanced Python Made Simple 2026

def greet():
    print("Hello!")

def add_excitement(func):
    """A decorator that adds excitement to any function."""
    def wrapper():
        print(">>> Starting...")
        func()
        print("<<< Done!")
    return wrapper

# Manually wrap the function
excited_greet = add_excitement(greet)
excited_greet()
# >>> Starting...
# Hello!
# <<< Done!

@add_excitement
def greet():
    print("Hello!")

# This is EXACTLY the same as:
# greet = add_excitement(greet)

greet()
# >>> Starting...
# Hello!
# <<< Done!

import functools

def add_excitement(func):
    @functools.wraps(func)  # Preserves func's __name__, __doc__, etc.
    def wrapper(*args, **kwargs):
        print(">>> Starting...")
        result = func(*args, **kwargs)
        print("<<< Done!")
        return result
    return wrapper

import time
import functools

def timer(func):
    """Measure and print function execution time."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f"{func.__name__} took {end - start:.4f}s")
        return result
    return wrapper

@timer
def process_data(n: int) -> list:
    return [i ** 2 for i in range(n)]

result = process_data(1_000_000)
# process_data took 0.1234s

import functools
import time

def retry(max_attempts: int = 3, delay: float = 1.0, exceptions=(Exception,)):
    """Retry a function on failure with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts:
                        wait = delay * (2 ** (attempt - 1))  # Exponential backoff
                        print(f"Attempt {attempt} failed: {e}. Retrying in {wait:.1f}s...")
                        time.sleep(wait)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, delay=1.0, exceptions=(ConnectionError, TimeoutError))
def fetch_from_api(url: str) -> dict:
    import requests
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

import functools

def memoize(func):
    """Cache function results for the same arguments."""
    cache = {}
    
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
            print(f"Computing {func.__name__}{args}...")
        else:
            print(f"Cache hit for {func.__name__}{args}")
        return cache[args]
    
    wrapper.cache = cache  # Expose cache for inspection
    return wrapper

@memoize
def fibonacci(n: int) -> int:
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(10))  # Computes recursively and caches each result
print(fibonacci(10))  # Instant — from cache

# Python's built-in: functools.lru_cache
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
    return sum(range(n))

import functools
from typing import Callable, TypeVar

def validate_positive(*arg_names):
    """Raise ValueError if specified arguments are not positive."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import inspect
            sig = inspect.signature(func)
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()
            
            for name in arg_names:
                if name in bound.arguments and bound.arguments[name] <= 0:
                    raise ValueError(f"Argument '{name}' must be positive, got {bound.arguments[name]}")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_positive("width", "height")
def create_rectangle(width: float, height: float) -> dict:
    return {"width": width, "height": height, "area": width * height}

create_rectangle(5, 3)   # Works fine
create_rectangle(-1, 3)  # Raises ValueError: Argument 'width' must be positive

@timer
@retry(max_attempts=3)
@validate_positive("n")
def process(n: int) -> list:
    return [i ** 2 for i in range(n)]

# Equivalent to: process = timer(retry(3)(validate_positive("n")(process)))
# Decorators apply bottom-up

# This loads ALL 1 million numbers into memory immediately
million_squares = [x ** 2 for x in range(1_000_000)]
# Uses ~8 MB of RAM

# This generates each number ON DEMAND — near zero memory
million_squares_gen = (x ** 2 for x in range(1_000_000))
# Uses ~112 bytes regardless of size

# You can still iterate over the generator the same way
for square in million_squares_gen:
    if square > 100:
        break  # Stop early without generating all 1M values

def countdown(n: int):
    print(f"Starting countdown from {n}")
    while n > 0:
        yield n
        n -= 1
    print("Blastoff!")

gen = countdown(3)  # Nothing runs yet
print(next(gen))    # "Starting countdown from 3" then yields 3
print(next(gen))    # yields 2
print(next(gen))    # yields 1
# next(gen)         # Would print "Blastoff!" then raise StopIteration

# for loops handle StopIteration automatically
for n in countdown(5):
    print(n)

def read_large_csv(filepath: str, batch_size: int = 1000):
    """Read a huge CSV file in batches without loading it all into memory."""
    import csv
    
    with open(filepath, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        batch = []
        
        for row in reader:
            batch.append(row)
            if len(batch) >= batch_size:
                yield batch
                batch = []
        
        if batch:
            yield batch  # Don't forget the last partial batch

# Process a 10GB CSV file with constant memory usage
for batch in read_large_csv("huge_dataset.csv", batch_size=500):
    for record in batch:
        process_record(record)  # Your processing function
    print(f"Processed batch of {len(batch)} records")

def fibonacci():
    """Generate Fibonacci numbers indefinitely."""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

def take(n: int, gen):
    """Take only the first n values from a generator."""
    for i, value in enumerate(gen):
        if i >= n:
            break
        yield value

# Get first 15 Fibonacci numbers without storing all of them
first_15 = list(take(15, fibonacci()))
print(first_15)
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377]

def read_logs(filepath: str):
    """Read log file line by line."""
    with open(filepath) as f:
        for line in f:
            yield line.strip()

def filter_errors(lines):
    """Keep only error lines."""
    for line in lines:
        if "ERROR" in line or "CRITICAL" in line:
            yield line

def parse_log(lines):
    """Parse each line into a dict."""
    import re
    pattern = r"(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}) (\w+) (.+)"
    for line in lines:
        match = re.match(pattern, line)
        if match:
            yield {
                "date": match.group(1),
                "time": match.group(2),
                "level": match.group(3),
                "message": match.group(4),
            }

def save_to_db(records):
    """Save each record to database."""
    for record in records:
        insert_into_db(record)  # Your DB insert function
        yield record

# Compose the pipeline — no data is loaded until iteration starts
pipeline = save_to_db(
    parse_log(
        filter_errors(
            read_logs("server.log")
        )
    )
)

# Process one record at a time — constant memory
for record in pipeline:
    pass
print("Pipeline complete")

# List comprehension — eager, loads all into memory
squares_list = [x**2 for x in range(10)]         # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# Generator expression — lazy, no memory
squares_gen = (x**2 for x in range(10))           # <generator object>

# Use generator when you only need to iterate once
total = sum(x**2 for x in range(1_000_000))       # Memory efficient!
first_over_100 = next(x for x in squares_gen if x > 100)

# Use list when you need to:
# - Access by index: squares[5]
# - Iterate multiple times
# - Check length: len(squares)
# - Sort or reverse

import functools
import time

def timed_generator(func):
    """Time how long a generator takes to complete."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        yield from func(*args, **kwargs)
        end = time.perf_counter()
        print(f"{func.__name__} completed in {end - start:.3f}s")
    return wrapper

@timed_generator
def process_items(items):
    for item in items:
        yield transform(item)  # Your transform function