Decorators & Context Managers

# The concept without decorator syntax
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("Before the function")
        result = func(*args, **kwargs)
        print("After the function")
        return result
    return wrapper

def greet(name):
    print(f"Hello, {name}!")

# Apply decorator manually
greet = my_decorator(greet)
greet("Alice")
# Before the function
# Hello, Alice!
# After the function

# The @ syntax does exactly the same thing
@my_decorator
def greet(name):
    print(f"Hello, {name}!")

import time
import functools

def timer(func):
    @functools.wraps(func)  # Preserves the original function's metadata
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(0.5)
    return "done"

@timer
def fast_function(n):
    return sum(range(n))

slow_function()           # slow_function took 0.5001s
print(fast_function(1000000))  # fast_function took 0.0234s

def repeat(times):
    """Repeat a function n times."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(3)
def say_hello():
    print("Hello!")

say_hello()
# Hello!
# Hello!
# Hello!

# Retry decorator — extremely useful for network calls
def retry(max_attempts=3, exceptions=(Exception,), delay=1.0):
    """Retry a function on failure."""
    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:
                        print(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...")
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, exceptions=(ConnectionError,), delay=2.0)
def fetch_data(url):
    # Might fail on flaky network
    import requests
    return requests.get(url, timeout=5).json()

# 1. Caching (memoization)
from functools import lru_cache

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

print(fibonacci(100))   # Instant after first call

# Manual cache decorator
def cache(func):
    memo = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args not in memo:
            memo[args] = func(*args)
        return memo[args]
    wrapper.cache = memo  # Expose the cache
    return wrapper

# 2. Logging decorator
import logging

def log_calls(logger=None, level=logging.INFO):
    if logger is None:
        logger = logging.getLogger(__name__)
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            logger.log(level, f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
            try:
                result = func(*args, **kwargs)
                logger.log(level, f"{func.__name__} returned {result!r}")
                return result
            except Exception as e:
                logger.error(f"{func.__name__} raised {type(e).__name__}: {e}")
                raise
        return wrapper
    return decorator

@log_calls()
def process_order(order_id, quantity):
    return {"order_id": order_id, "status": "processed", "qty": quantity}

# 3. Authentication / Authorization
def require_auth(func):
    @functools.wraps(func)
    def wrapper(request, *args, **kwargs):
        if not getattr(request, 'user', None) or not request.user.is_authenticated:
            raise PermissionError("Authentication required")
        return func(request, *args, **kwargs)
    return wrapper

def require_permission(permission):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(request, *args, **kwargs):
            if not request.user.has_permission(permission):
                raise PermissionError(f"Permission '{permission}' required")
            return func(request, *args, **kwargs)
        return wrapper
    return decorator

# 4. Rate limiting
def rate_limit(calls_per_second):
    import threading
    lock = threading.Lock()
    call_times = []
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with lock:
                now = time.time()
                call_times[:] = [t for t in call_times if now - t < 1.0]
                if len(call_times) >= calls_per_second:
                    raise RuntimeError(f"Rate limit: max {calls_per_second} calls/second")
                call_times.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

def singleton(cls):
    """Ensure only one instance of a class exists."""
    instances = {}
    
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    
    return get_instance

@singleton
class DatabasePool:
    def __init__(self):
        self.connections = []
        print("Creating database pool")

db1 = DatabasePool()
db2 = DatabasePool()
print(db1 is db2)  # True — same instance

# Adding methods to a class
def add_method(cls):
    def new_method(self):
        return f"Added to {cls.__name__}"
    cls.added_method = new_method
    return cls

@add_method
class MyClass:
    pass

obj = MyClass()
print(obj.added_method())

# Using contextlib.contextmanager — the easiest way
from contextlib import contextmanager, suppress

@contextmanager
def managed_resource(name):
    print(f"Setting up {name}")
    resource = {"name": name, "active": True}
    try:
        yield resource  # Execution pauses here; 'as' variable gets resource
    except Exception as e:
        print(f"Exception in {name}: {e}")
        raise
    finally:
        resource["active"] = False
        print(f"Cleaning up {name}")

with managed_resource("database") as db:
    print(f"Using resource: {db['name']}")
    # Even if exception here, cleanup runs

# suppress: ignore specific exceptions
with suppress(FileNotFoundError):
    import os
    os.remove("nonexistent.txt")  # No error raised

# ExitStack: manage multiple resources dynamically
from contextlib import ExitStack

files = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
    handles = [stack.enter_context(open(f, 'w')) for f in files]
    for handle in handles:
        handle.write("data\n")
# All files closed automatically

@timer
@retry(max_attempts=3)
@log_calls()
def fetch_user_data(user_id):
    # Execution order: log → retry → timer → function
    return {"id": user_id, "name": "Alice"}

# Equivalent to:
# fetch_user_data = timer(retry(3)(log_calls()(fetch_user_data)))
# Decorators apply from bottom to top

def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper  # wrapper.__name__ = 'wrapper', not the original!

def good_decorator(func):
    @functools.wraps(func)  # Copies name, docstring, annotations
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def my_function():
    """My docstring."""
    pass

@good_decorator
def my_function2():
    """My docstring."""
    pass

print(my_function.__name__)   # 'wrapper' — broken!
print(my_function2.__name__)  # 'my_function2' — correct
print(my_function2.__doc__)   # 'My docstring.' — preserved