Higher Order Functions — Decorator Ordering Auth Bypass

# Package: io.thecodeforge.python.functional
# Demonstrates: first-class functions, closures, factory pattern,
# and the classic loop-closure variable capture trap

def double(x): return x * 2
def square(x): return x ** 2
def negate(x): return -x

# Functions are objects — assign to a variable just like any other value
transform = double
print(transform(5))  # 10 — calling via the variable

# Pass a function as an argument to another function
# apply_to_list is a higher order function: it accepts a function as a parameter
def apply_to_list(func, lst):
    return [func(x) for x in lst]

numbers = [1, 2, 3, 4, 5]
print(apply_to_list(double, numbers))   # [2, 4, 6, 8, 10]
print(apply_to_list(square, numbers))   # [1, 4, 9, 16, 25]
print(apply_to_list(negate, numbers))   # [-1, -2, -3, -4, -5]

# Return a function from a function — this is the factory pattern
# make_multiplier is a higher order function: it returns a function
def make_multiplier(factor):
    def multiplier(x):
        return x * factor  # 'factor' is captured from the enclosing scope
    return multiplier

times3 = make_multiplier(3)
times10 = make_multiplier(10)
print(times3(7))   # 21  — factor=3 is captured in the closure
print(times10(7))  # 70  — factor=10 is captured separately

# TRAP: loop closures capture the variable, not the value
# All three functions will return 2 — the final value of i
funcs_broken = [lambda: i for i in range(3)]
print([f() for f in funcs_broken])  # [2, 2, 2] — NOT [0, 1, 2]

# FIX 1: default argument binding captures the current value at creation time
funcs_fixed = [lambda i=i: i for i in range(3)]
print([f() for f in funcs_fixed])   # [0, 1, 2] — correct

# FIX 2: functools.partial — cleaner for named functions
import functools

def add(x, y): return x + y

adders = [functools.partial(add, i) for i in range(3)]
print([f(10) for f in adders])  # [10, 11, 12] — correct

# Inspect closure internals — useful for debugging
print(times3.__closure__[0].cell_contents)   # 3
print(times10.__closure__[0].cell_contents)  # 10

# Package: io.thecodeforge.python.functional
# Demonstrates: map, filter, reduce, iterator exhaustion trap,
# list comprehension equivalents, and when each is appropriate

from functools import reduce
import math

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# map: apply a function to every element
# Returns a lazy iterator in Python 3 — wrap in list() to materialize
doubled = list(map(lambda x: x * 2, numbers))
print(doubled)  # [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]

# filter: keep elements where the function returns True
# Also returns a lazy iterator in Python 3
evens = list(filter(lambda x: x % 2 == 0, numbers))
print(evens)    # [2, 4, 6, 8, 10]

# reduce: fold the list into a single value
# Removed from builtins in Python 3 — import from functools
total = reduce(lambda acc, x: acc + x, numbers)
print(total)    # 55

product = reduce(lambda acc, x: acc * x, numbers)
print(product)  # 3628800 (10!)

# TRAP: iterator exhaustion — the silent bug that never raises an exception
map_result = map(lambda x: x * 2, numbers)
first_pass = list(map_result)   # [2, 4, 6, ...] — works
second_pass = list(map_result)  # [] — silently empty, iterator is exhausted
print(f'First: {first_pass[:3]}, Second: {second_pass}')  # First: [2, 4, 6], Second: []

# List comprehensions: the Pythonic equivalent for simple cases
# More readable, eager (returns a list immediately), supports inline filtering
doubled_comp = [x * 2 for x in numbers]               # same as map example above
evens_comp   = [x for x in numbers if x % 2 == 0]     # same as filter example above
both         = [x * 2 for x in numbers if x % 2 == 0] # filter AND map in one expression
print(both)  # [4, 8, 12, 16, 20]

# When map() wins: named built-in functions with C implementations
# map(math.sqrt, numbers) is faster than [math.sqrt(x) for x in numbers]
# because there's no per-element Python overhead for the function call
sqrt_all = list(map(math.sqrt, numbers))
print([round(v, 3) for v in sqrt_all])  # [1.0, 1.414, 1.732, 2.0, 2.236, ...]

# When reduce() is appropriate: composing a pipeline of functions
# Each function takes the output of the previous one
def add_tax(price):     return round(price * 1.08, 2)
def apply_discount(price): return round(price * 0.9, 2)
def round_to_cent(price):  return round(price, 2)

pipeline = [add_tax, apply_discount, round_to_cent]
base_price = 100.0
final_price = reduce(lambda value, fn: fn(value), pipeline, base_price)
print(f'Final price after pipeline: ${final_price}')  # $97.2

# Don't use reduce() for simple sums — use sum() which handles empty lists
print(sum(numbers))         # 55 — cleaner, no initializer needed
print(sum([]))              # 0  — handles empty sequence gracefully
print(math.prod(numbers))  # 3628800 — same as reduce product above

# Package: io.thecodeforge.python.functional
# Demonstrates: simple decorator, decorator factory, class-based decorator,
# @wraps importance, and decorator stacking order

import time
import functools
from typing import Callable, TypeVar, Any

F = TypeVar('F', bound=Callable[..., Any])


# --- Simple decorator ---

def timer(func: F) -> F:
    """Decorator that prints execution time. Always use @wraps."""
    @functools.wraps(func)  # copies __name__, __doc__, __module__, __qualname__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed_ms = (time.perf_counter() - start) * 1000
        print(f'{func.__name__} took {elapsed_ms:.2f}ms')
        return result
    return wrapper  # type: ignore[return-value]


@timer
def slow_sum(n: int) -> int:
    """Sum integers from 0 to n-1."""
    return sum(range(n))


result = slow_sum(1_000_000)
print(f'Result: {result}')
print(f'Function name preserved: {slow_sum.__name__}')  # 'slow_sum', not 'wrapper'
print(f'Docstring preserved: {slow_sum.__doc__}')        # 'Sum integers from 0 to n-1.'


# --- Decorator factory: decorator that takes parameters ---
# Three levels of nesting: factory -> decorator -> wrapper

def retry(max_attempts: int = 3, delay_seconds: float = 0.1):
    """Decorator factory: retries the decorated function on exception."""
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as exc:
                    last_exc = exc
                    if attempt < max_attempts:
                        print(f'{func.__name__} attempt {attempt} failed: {exc}. Retrying...')
                        time.sleep(delay_seconds)
            raise RuntimeError(
                f'{func.__name__} failed after {max_attempts} attempts'
            ) from last_exc
        return wrapper  # type: ignore[return-value]
    return decorator


@retry(max_attempts=3, delay_seconds=0.05)
def flaky_api_call(fail_count: list) -> str:
    """Simulates an API call that fails the first N times."""
    if fail_count:
        fail_count.pop()
        raise ConnectionError('Simulated network error')
    return 'success'


fails = [1, 2]  # will fail twice, succeed on third attempt
print(flaky_api_call(fails))  # 'success' after two retries


# --- Class-based decorator: stateful across calls ---
# Use when you need to maintain state (call count, cache, rate limit window)

class RateLimit:
    """Class-based decorator: limits a function to max_calls per window_seconds."""

    def __init__(self, max_calls: int, window_seconds: float):
        self.max_calls = max_calls
        self.window_seconds = window_seconds
        self.calls: list[float] = []

    def __call__(self, func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            # Evict calls outside the current window
            self.calls = [t for t in self.calls if now - t < self.window_seconds]
            if len(self.calls) >= self.max_calls:
                raise RuntimeError(
                    f'Rate limit exceeded: {self.max_calls} calls '
                    f'per {self.window_seconds}s'
                )
            self.calls.append(now)
            return func(*args, **kwargs)
        return wrapper  # type: ignore[return-value]

    # __get__ makes this work correctly as a method decorator
    # Without it, the class-based decorator breaks on instance methods
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return functools.partial(self, obj)


@RateLimit(max_calls=3, window_seconds=1.0)
def send_notification(message: str) -> None:
    print(f'Sending: {message}')


for i in range(3):
    send_notification(f'Message {i}')
# send_notification('Message 4')  # would raise RuntimeError: Rate limit exceeded


# --- Decorator stacking order demonstration ---
# Decorators apply bottom-up (inner first), execute top-down (outer first)

def log_call(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f'[LOG] Calling {func.__name__}')
        result = func(*args, **kwargs)
        print(f'[LOG] {func.__name__} returned {result}')
        return result
    return wrapper

def validate_positive(func):
    @functools.wraps(func)
    def wrapper(n, *args, **kwargs):
        if n < 0:
            raise ValueError(f'{func.__name__} requires a positive integer, got {n}')
        return func(n, *args, **kwargs)
    return wrapper


# @log_call runs first (outer), @validate_positive runs second (inner)
# Execution order: log_call wrapper -> validate_positive wrapper -> double_it
@log_call
@validate_positive
def double_it(n: int) -> int:
    return n * 2


double_it(5)
# [LOG] Calling double_it
# [LOG] double_it returned 10

# Package: io.thecodeforge.python.functional
# Demonstrates: functools.partial, lru_cache, cache, singledispatch
# These are the functools utilities you'll actually use in production

import functools
import math
import time
from typing import Any


# --- functools.partial: freeze arguments, produce a new callable ---
# Cleaner than a lambda, and picklable (lambdas are not)

def power(base: float, exponent: float) -> float:
    return base ** exponent

# Freeze the exponent, produce specialized functions
square = functools.partial(power, exponent=2)
cube   = functools.partial(power, exponent=3)

print(square(5))   # 25.0
print(cube(3))     # 27.0

# In multiprocessing, use partial instead of lambda — lambdas cannot be pickled
# pool.map(functools.partial(process_record, config=cfg), records)  # works
# pool.map(lambda r: process_record(r, config=cfg), records)        # PicklingError


# --- functools.lru_cache: memoize expensive function calls ---
# maxsize=None is equivalent to functools.cache (Python 3.9+)
# Arguments must be hashable — lists and dicts are not

@functools.lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    """Fibonacci with memoization — O(n) instead of O(2^n)."""
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

start = time.perf_counter()
print(fibonacci(40))                        # 102334155
print(f'Cached calls: {fibonacci.cache_info()}')  # shows hits vs misses
print(f'Time: {(time.perf_counter()-start)*1000:.2f}ms')

# functools.cache: same as lru_cache(maxsize=None), Python 3.9+
@functools.cache
def expensive_lookup(key: str) -> str:
    """Simulates a database lookup — cached after first call per key."""
    time.sleep(0.001)  # simulate I/O
    return f'value_for_{key}'

print(expensive_lookup('user_123'))  # slow first call
print(expensive_lookup('user_123'))  # instant — from cache


# --- functools.singledispatch: type-based function overloading ---
# Useful in serialization, rendering, and validation pipelines
# where behavior varies by input type

@functools.singledispatch
def serialize(value: Any) -> str:
    """Default serializer — fallback for unregistered types."""
    raise TypeError(f'No serializer registered for type {type(value).__name__}')

@serialize.register(int)
@serialize.register(float)
def _(value: float) -> str:
    return f'number:{value}'

@serialize.register(str)
def _(value: str) -> str:
    return f'string:{value!r}'

@serialize.register(list)
def _(value: list) -> str:
    return f'list:[{', '.join(serialize(item) for item in value)}]'

@serialize.register(bool)  # register before int — bool is a subclass of int
def _(value: bool) -> str:
    return f'bool:{'true' if value else 'false'}'


print(serialize(42))              # number:42
print(serialize('hello'))         # string:'hello'
print(serialize([1, 'two', 3.0])) # list:[number:1, string:'two', number:3.0]
print(serialize(True))            # bool:true


# --- functools.reduce with function pipeline pattern ---
# The one use case where reduce is genuinely expressive

def compose(*functions):
    """Compose functions right-to-left: compose(f, g, h)(x) = f(g(h(x)))"""
    return functools.reduce(lambda f, g: lambda *args: f(g(*args)), functions)


process = compose(
    lambda x: x ** 2,
    lambda x: x + 1,
    lambda x: x * 2
)
# process(3) = (3*2 + 1)^2 = 49
print(process(3))  # 49

# io.thecodeforge.hof-callable-class

from typing import Iterable

class SquareWithLog:
    def __init__(self, label: str):
        self.label = label
        self.count = 0

    def __call__(self, n: int) -> int:
        self.count += 1
        result = n * n
        print(f"[{self.label}] square({n}) = {result}")
        return result

squarer = SquareWithLog("worker")
result = list(map(squarer, [1, 2, 3, 4]))
print(f"Results: {result}")
print(f"Total calls: {squarer.count}")

# io.thecodeforge.hof-partial

from functools import partial

def scale(data: list, factor: float, offset: float) -> list:
    return [x * factor + offset for x in data]

# Factory using partial — zero closures, pure intent
normalize = partial(scale, factor=1.0, offset=0.0)
standardize = partial(scale, factor=0.5, offset=-1.0)

print(normalize([10, 20, 30]))
print(standardize([10, 20, 30]))

# Inspect the partially-applied function
print(f"Function: {standardize.func.__name__}")
print(f"Fixed args: {standardize.keywords}")

# io.thecodeforge.hof-singledispatch

from functools import singledispatch

@singledispatch
def serialize(obj):
    raise TypeError(f"Unsupported type: {type(obj)}")

@serialize.register(int)
def _serialize_int(n):
    return f"int:{n}"

@serialize.register(list)
def _serialize_list(items):
    return f"list:[{','.join(str(x) for x in items)}]"

@serialize.register(dict)
def _serialize_dict(d):
    pairs = ','.join(f"{k}={v}" for k, v in d.items())
    return f"dict:{{{pairs}}}"

print(serialize(42))
print(serialize([1, 2, 3]))
print(serialize({"a": 1, "b": 2}))