Closures & Decorators¶

def make_adder(n):
    def adder(x):
        return x + n       # 'n' is a free variable
    return adder

add5  = make_adder(5)
add10 = make_adder(10)
print(add5(3))    # 8
print(add10(3))   # 13

# Each closure has its own cell holding a different 'n'
print(add5.__closure__)           # (<cell at 0x...>,)
print(add5.__closure__[0].cell_contents)    # 5
print(add10.__closure__[0].cell_contents)   # 10

# Cell objects are shared: mutating nonlocal updates the cell
def counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc

c = counter()
print(c(), c(), c())   # 1 2 3
# Inspect: the cell still holds the latest count
print(c.__closure__[0].cell_contents)  # 3

import time
import functools

def timer(fn):
    @functools.wraps(fn)       # preserves __name__, __doc__, __annotations__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{fn.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_sum(n):
    """Sum integers 0..n."""
    return sum(range(n))

print(slow_sum(10_000_000))
# slow_sum took 0.3211s
# 49999995000000

# The @ syntax is purely shorthand:
# @timer         is identical to:
# slow_sum = timer(slow_sum)

# functools.wraps matters:
print(slow_sum.__name__)   # 'slow_sum'  (not 'wrapper'!)
print(slow_sum.__doc__)    # 'Sum integers 0..n.'

import time, functools, random

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    """Decorator factory: retry on specified exceptions."""
    def decorator(fn):                      # level 2 – receives the function
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):       # level 3 – called at runtime
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as exc:
                    last_exc = exc
                    print(f"Attempt {attempt}/{max_attempts} failed: {exc}")
                    if attempt < max_attempts:
                        time.sleep(delay * attempt)   # exponential-ish backoff
            raise last_exc
        return wrapper
    return decorator     # decorator factory returns decorator

# Usage
@retry(max_attempts=4, delay=0.1, exceptions=(ValueError, IOError))
def flaky_api_call(endpoint):
    if random.random() < 0.7:
        raise ValueError(f"Transient error on {endpoint}")
    return f"Success from {endpoint}"

random.seed(99)
print(flaky_api_call("/health"))

import functools, time

class RateLimiter:
    """Allow at most `calls` calls per `period` seconds."""
    def __init__(self, calls, period):
        self.calls  = calls
        self.period = period
        self.history = []

    def __call__(self, fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            # Purge old entries outside the window
            self.history = [t for t in self.history if now - t < self.period]
            if len(self.history) >= self.calls:
                raise RuntimeError(
                    f"Rate limit exceeded: {self.calls} calls/{self.period}s"
                )
            self.history.append(now)
            return fn(*args, **kwargs)
        return wrapper

@RateLimiter(calls=3, period=1.0)
def fetch_quote(symbol):
    return f"Price of {symbol}: ${round(100 + hash(symbol) % 50, 2)}"

for _ in range(3):
    print(fetch_quote("AAPL"))

try:
    fetch_quote("AAPL")   # 4th call within 1 second
except RuntimeError as e:
    print(f"Caught: {e}")

import functools

def bold(fn):
    @functools.wraps(fn)
    def wrapper(*a, **kw): return f"<b>{fn(*a, **kw)}</b>"
    return wrapper

def italic(fn):
    @functools.wraps(fn)
    def wrapper(*a, **kw): return f"<i>{fn(*a, **kw)}</i>"
    return wrapper

@bold       # applied second (outer)
@italic     # applied first (inner)
def greet(name):
    return f"Hello, {name}"

print(greet("Alice"))   # <b><i>Hello, Alice</i></b>

# Execution order reads bottom-up for application,
# top-down for actual invocation:
# greet = bold(italic(greet))
# call: bold.wrapper → italic.wrapper → original greet

import functools

# @cache (Python 3.9+) is lru_cache(maxsize=None) – unbounded
@functools.cache
def fib(n):
    if n < 2: return n
    return fib(n-1) + fib(n-2)

print(fib(50))            # 12586269025  (instant)
print(fib.cache_info())   # hits=..., misses=51, maxsize=None, currsize=51

# @lru_cache(maxsize=N) – bounded LRU eviction
@functools.lru_cache(maxsize=128)
def heavy_query(region, year):
    # simulate DB round-trip
    import time; time.sleep(0.01)
    return f"data_{region}_{year}"

heavy_query("APAC", 2023)
heavy_query("APAC", 2023)   # cache hit
print(heavy_query.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

heavy_query.cache_clear()   # manual invalidation

class Temperature:
    def __init__(self, celsius):
        self._celsius = celsius

    @property
    def celsius(self):
        """Getter – accessed like an attribute."""
        return self._celsius

    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError(f"Below absolute zero: {value}")
        self._celsius = value

    @celsius.deleter
    def celsius(self):
        del self._celsius

    @property
    def fahrenheit(self):
        return self._celsius * 9/5 + 32

    @classmethod
    def from_fahrenheit(cls, f):
        """Alternative constructor – receives the class, not an instance."""
        return cls((f - 32) * 5/9)

    @staticmethod
    def is_valid(value):
        """Utility with no access to class or instance."""
        return value >= -273.15

t = Temperature(100)
print(t.celsius)       # 100
print(t.fahrenheit)    # 212.0
t.celsius = 37
print(t.celsius)       # 37

body = Temperature.from_fahrenheit(98.6)
print(round(body.celsius, 1))   # 37.0

print(Temperature.is_valid(-300))   # False