Decorators, Generators & Context Managers

# Step 1: A decorator is just a function that takes a function
def shout(func):
    """A decorator that converts the result to uppercase."""
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return result.upper()
    return wrapper

# Step 2: Apply it manually (the old way)
def greet(name):
    return f"Hello, {name}!"

greet = shout(greet)  # wrap greet with shout
print(greet("Alice"))  # HELLO, ALICE!

# Step 3: Apply it with @ syntax (the modern way -- identical result)
@shout
def greet_v2(name):
    return f"Hello, {name}!"

print(greet_v2("Bob"))  # HELLO, BOB!

import time

def timer(func):
    """Measure how long a function takes to run."""
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.4f} seconds")
        return result
    return wrapper

@timer
def slow_sum(n):
    """Sum numbers from 0 to n using a loop."""
    total = 0
    for i in range(n):
        total += i
    return total

@timer
def fast_sum(n):
    """Sum numbers from 0 to n using the formula."""
    return n * (n - 1) // 2

slow_sum(1_000_000)   # slow_sum took 0.0523 seconds
fast_sum(1_000_000)   # fast_sum took 0.0000 seconds

def log_call(func):
    """Log every call to a function with its arguments."""
    def wrapper(*args, **kwargs):
        args_str = ", ".join([repr(a) for a in args])
        kwargs_str = ", ".join([f"{k}={v!r}" for k, v in kwargs.items()])
        all_args = ", ".join(filter(None, [args_str, kwargs_str]))
        print(f"CALL: {func.__name__}({all_args})")
        result = func(*args, **kwargs)
        print(f"  -> returned {result!r}")
        return result
    return wrapper

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

@log_call
def multiply(a, b):
    return a * b

add(3, 5)
# CALL: add(3, 5)
#   -> returned 8
multiply(4, 7)
# CALL: multiply(4, 7)
#   -> returned 28

def cache(func):
    """Cache results to avoid recomputing for the same inputs."""
    memo = {}
    def wrapper(*args):
        if args in memo:
            print(f"  (cache hit for {args})")
            return memo[args]
        result = func(*args)
        memo[args] = result
        return result
    return wrapper

@cache
def fibonacci(n):
    """Compute the nth Fibonacci number recursively."""
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(10))  # 55 (fast -- only computes each value once)
print(fibonacci(10))  # 55 (instant -- cache hit)

import time
from functools import wraps

# Pattern: outer function takes arguments, returns the real decorator
def retry(max_retries: int = 3, delay: float = 1.0):
    """Decorator factory: retry a function on exception up to max_retries times."""
    def decorator(func):          # the real decorator — takes the function
        @wraps(func)              # always use @wraps to preserve metadata
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"  Attempt {attempt}/{max_retries} failed: {e}")
                    if attempt < max_retries:
                        time.sleep(delay)
            raise RuntimeError(f"{func.__name__} failed after {max_retries} attempts")
        return wrapper
    return decorator              # return the decorator, not the wrapped function

# Usage: @retry(max_retries=4, delay=0.5) calls retry(4, 0.5), which returns decorator
@retry(max_retries=4, delay=0.1)
def unstable_api_call():
    """Simulates an API that sometimes fails."""
    import random
    if random.random() < 0.7:
        raise ConnectionError("Server busy")
    return "OK"

# Without arguments uses defaults: @retry is equivalent to @retry()
@retry()
def another_fn():
    return 42

result = another_fn()   # never retries since it always succeeds

def count_up(n):
    """A generator that yields numbers from 1 to n."""
    i = 1
    while i <= n:
        yield i
        i += 1

# Using the generator
counter = count_up(5)
print(next(counter))  # 1
print(next(counter))  # 2
print(next(counter))  # 3

# Or use it in a for loop (most common)
for num in count_up(5):
    print(num, end=" ")
# 1 2 3 4 5

import sys

# List: stores ALL values in memory at once
numbers_list = [x ** 2 for x in range(1_000_000)]
print(f"List size: {sys.getsizeof(numbers_list):,} bytes")
# List size: 8,448,728 bytes (~8 MB)

# Generator: produces values one at a time
numbers_gen = (x ** 2 for x in range(1_000_000))
print(f"Generator size: {sys.getsizeof(numbers_gen):,} bytes")
# Generator size: 200 bytes (constant, regardless of how many items!)

def fibonacci_gen():
    """An infinite Fibonacci generator."""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# Get the first 10 Fibonacci numbers
fib = fibonacci_gen()
first_10 = [next(fib) for _ in range(10)]
print(first_10)  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

# Without yield from: manual delegation
def chain_manual(gen1, gen2):
    for item in gen1:
        yield item
    for item in gen2:
        yield item

# With yield from: delegate directly
def chain(gen1, gen2):
    yield from gen1   # yield every value from gen1, then...
    yield from gen2   # yield every value from gen2

gen = chain(range(3), range(10, 13))
print(list(gen))  # [0, 1, 2, 10, 11, 12]


# yield from works with any iterable, including lists and strings
def flatten(nested):
    """Recursively flatten arbitrarily nested lists using yield from."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)  # recurse and re-yield all values
        else:
            yield item

print(list(flatten([1, [2, [3, 4]], [5, 6]])))  # [1, 2, 3, 4, 5, 6]

def read_lines(filename):
    """Yield lines from a file one at a time."""
    with open(filename) as f:
        for line in f:
            yield line.strip()

def filter_nonempty(lines):
    """Yield only non-empty lines."""
    for line in lines:
        if line:
            yield line

def to_uppercase(lines):
    """Yield each line in uppercase."""
    for line in lines:
        yield line.upper()

# Chain generators into a pipeline (nothing executes until you iterate!)
# pipeline = to_uppercase(filter_nonempty(read_lines("data.txt")))
# for line in pipeline:
#     print(line)

class Timer:
    """A context manager that times a code block."""

    def __enter__(self):
        import time
        self.start = time.time()
        print("Timer started...")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        elapsed = time.time() - self.start
        print(f"Timer stopped: {elapsed:.4f} seconds")
        return False  # do not suppress exceptions

# Usage
with Timer():
    total = sum(range(1_000_000))
    print(f"Sum: {total}")
# Timer started...
# Sum: 499999500000
# Timer stopped: 0.0312 seconds

class DatabaseConnection:
    """Simulated database connection with automatic cleanup."""

    def __init__(self, db_name):
        self.db_name = db_name
        self.connected = False

    def __enter__(self):
        print(f"Connecting to {self.db_name}...")
        self.connected = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"Closing connection to {self.db_name}")
        self.connected = False
        if exc_type:
            print(f"  (error occurred: {exc_val})")
        return False

    def query(self, sql):
        if not self.connected:
            raise RuntimeError("Not connected!")
        print(f"  Executing: {sql}")
        return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]


# The connection is guaranteed to close, even if query fails
with DatabaseConnection("ml_data.db") as db:
    results = db.query("SELECT * FROM users")
    print(f"  Got {len(results)} results")
# Connecting to ml_data.db...
#   Executing: SELECT * FROM users
#   Got 2 results
# Closing connection to ml_data.db

from contextlib import contextmanager
import time

@contextmanager
def timer():
    """A timer context manager using contextlib."""
    start = time.time()
    print("Timer started...")
    try:
        yield  # this is where the 'with' block runs
    finally:
        # try/finally is essential — without it, an exception inside the
        # `with` block skips the cleanup code below `yield`.
        elapsed = time.time() - start
        print(f"Timer stopped: {elapsed:.4f} seconds")

with timer():
    total = sum(range(1_000_000))
    print(f"Sum: {total}")

import time
from functools import wraps

def smart_cache(func):
    """A decorator that caches results and logs timing."""
    memo = {}

    @wraps(func)  # preserves the original function's name and docstring
    def wrapper(*args):
        if args in memo:
            print(f"  {func.__name__}{args}: cache hit")
            return memo[args]

        start = time.time()
        result = func(*args)
        elapsed = time.time() - start

        memo[args] = result
        print(f"  {func.__name__}{args}: computed in {elapsed:.6f}s")
        return result

    return wrapper

@smart_cache
def expensive_calculation(n):
    """Simulate an expensive computation."""
    total = 0
    for i in range(n * 1000):
        total += i ** 0.5
    return total

print(expensive_calculation(100))   # computed in 0.012345s
print(expensive_calculation(100))   # cache hit (instant)
print(expensive_calculation(200))   # computed in 0.024567s