functools @lru_cache — Cross-User Cache Leak

import functools
import time

# Without caching this naive recursive fibonacci is catastrophically slow.
# fib(40) makes over 300 million recursive calls.
@functools.lru_cache(maxsize=128)  # Cache up to 128 unique argument combinations
def fibonacci(n):
    """Return the nth Fibonacci number using memoized recursion."""
    if n < 2:
        return n
    # On the second call with the same n, this line never executes — cache takes over
    return fibonacci(n - 1) + fibonacci(n - 2)

# --- Demonstrating the speed difference ---
start = time.perf_counter()
result = fibonacci(50)  # Would take minutes without caching
elapsed = time.perf_counter() - start

print(f"fibonacci(50) = {result}")
print(f"Computed in {elapsed:.6f} seconds")

# lru_cache gives you a built-in stats tool — use it to verify caching is working
cache_info = fibonacci.cache_info()
print(f"Cache hits: {cache_info.hits}")    # How many times cache answered instead of the function
print(f"Cache misses: {cache_info.misses}") # How many times the function actually ran
print(f"Cache size: {cache_info.currsize}") # How many results are stored right now

# Call again with a cached value to see hits go up
_ = fibonacci(50)
print(f"\nAfter second call — Cache hits: {fibonacci.cache_info().hits}")

import functools

# A general-purpose logging function with many configurable parameters
def log_message(level, timestamp_format, application_name, message):
    """Write a formatted log line."""
    import datetime
    timestamp = datetime.datetime.now().strftime(timestamp_format)
    print(f"[{timestamp}] [{level}] [{application_name}] {message}")

# In our payments service we always use the same level, format, and app name.
# Instead of passing all four args every single time, we create a specialised version.
payments_logger = functools.partial(
    log_message,
    level="ERROR",                # Locked in — always ERROR for this logger
    timestamp_format="%H:%M:%S",  # Locked in — always this format
    application_name="PaymentsService"  # Locked in
)

# Now we only supply the one argument that actually changes: the message
payments_logger(message="Card charge failed — insufficient funds")
payments_logger(message="Refund processing timeout after 30s")

print()

# --- Another real-world use: adapting functions for map() ---
def apply_discount(price, discount_rate):
    """Return price after applying a percentage discount."""
    return round(price * (1 - discount_rate), 2)

# Black Friday — everything gets a 30% discount
# We lock in the discount_rate so map() only needs to supply price
black_friday_discount = functools.partial(apply_discount, discount_rate=0.30)

original_prices = [99.99, 149.99, 29.99, 199.99]
discounted_prices = list(map(black_friday_discount, original_prices))

print("Original prices: ", original_prices)
print("After 30% off:   ", discounted_prices)

# You can inspect what was locked in — great for debugging
print(f"\nLocked-in keywords: {black_friday_discount.keywords}")
print(f"Underlying function: {black_friday_discount.func.__name__}")

import functools
import operator

# ── PART 1: functools.wraps ──────────────────────────────────────────────────

def execution_timer(func):
    """A decorator that measures how long a function takes to run."""
    @functools.wraps(func)  # WITHOUT this, wrapper.__name__ would be 'wrapper', not the real name
    def wrapper(*args, **kwargs):
        import time
        start = time.perf_counter()
        result = func(*args, **kwargs)  # Call the original function
        elapsed = time.perf_counter() - start
        print(f"  ⏱  {func.__name__} completed in {elapsed:.4f}s")
        return result
    return wrapper

@execution_timer
def fetch_user_profile(user_id):
    """Simulate fetching a user profile from a database."""
    import time
    time.sleep(0.05)  # Simulated DB latency
    return {"id": user_id, "name": "Alex Rivera", "plan": "premium"}

profile = fetch_user_profile(user_id=42)

# Because we used @wraps, the function's true identity is intact
print(f"Function name:    {fetch_user_profile.__name__}")  # 'fetch_user_profile', not 'wrapper'
print(f"Docstring:        {fetch_user_profile.__doc__}")
print()

# ── PART 2: functools.reduce ─────────────────────────────────────────────────

# Real-world scenario: merge a list of permission dictionaries into one
# Each dict represents permissions granted by a different role
role_permissions = [
    {"read": True,  "write": False, "delete": False},
    {"read": True,  "write": True,  "delete": False},
    {"read": True,  "write": True,  "delete": True },
]

def merge_permissions(accumulated, new_role):
    """Union two permission dicts — True wins over False (most permissive merge)."""
    return {key: accumulated[key] or new_role[key] for key in accumulated}

# reduce applies merge_permissions left-to-right across the list
final_permissions = functools.reduce(merge_permissions, role_permissions)
print("Merged permissions:", final_permissions)

# Classic use: compute a product of a list (no built-in like sum() exists for this)
monthly_growth_rates = [1.05, 1.03, 1.07, 1.02]  # 5%, 3%, 7%, 2% monthly growth
cumulative_growth = functools.reduce(operator.mul, monthly_growth_rates, 1.0)
print(f"Cumulative growth over 4 months: {cumulative_growth:.4f}x")  # ~1.1837x

import functools

@functools.total_ordering  # Give us all comparison operators from just two methods
class SupportTicket:
    """
    A customer support ticket. Tickets are compared by priority first,
    then by creation time if priorities are equal.
    Lower priority number = higher urgency (1 is most critical).
    """
    PRIORITY_LABELS = {1: "Critical", 2: "High", 3: "Medium", 4: "Low"}

    def __init__(self, ticket_id, priority, created_at):
        self.ticket_id = ticket_id
        self.priority = priority    # 1 = most urgent
        self.created_at = created_at

    def __repr__(self):
        label = self.PRIORITY_LABELS[self.priority]
        return f"Ticket({self.ticket_id}, {label})"

    def __eq__(self, other):
        if not isinstance(other, SupportTicket):
            return NotImplemented  # Let Python handle comparison with other types gracefully
        return (self.priority, self.created_at) == (other.priority, other.created_at)

    def __lt__(self, other):
        if not isinstance(other, SupportTicket):
            return NotImplemented
        # Lower priority number = more urgent = comes first in sorted order
        return (self.priority, self.created_at) < (other.priority, other.created_at)

    # @total_ordering generates __le__, __gt__, __ge__ automatically from __eq__ and __lt__

import datetime

tickets = [
    SupportTicket("TKT-004", priority=3, created_at=datetime.datetime(2024, 6, 1, 10, 0)),
    SupportTicket("TKT-001", priority=1, created_at=datetime.datetime(2024, 6, 1,  9, 0)),
    SupportTicket("TKT-007", priority=2, created_at=datetime.datetime(2024, 6, 1, 11, 0)),
    SupportTicket("TKT-003", priority=1, created_at=datetime.datetime(2024, 6, 1,  8, 0)),
]

# sorted() works because total_ordering gave us all the operators we needed
priority_queue = sorted(tickets)
print("Tickets in priority order:")
for ticket in priority_queue:
    print(f"  {ticket}")

# The generated operators work correctly
print(f"\nTKT-001 > TKT-007?  {tickets[1] > tickets[2]}")  # True — priority 1 beats priority 2
print(f"TKT-004 >= TKT-007? {tickets[0] >= tickets[2]}")  # False — priority 3 is less urgent

import functools
import re
import time

class DataPipeline:
    def __init__(self, raw_data: list):
        self.raw_data = raw_data

    @functools.cached_property
    def cleaned_data(self):
        """Expensive cleaning operation: only computed when first accessed."""
        print("  [CLEANING] Running expensive data cleaning...")
        time.sleep(0.5)  # Simulate heavy processing
        # Remove None values, strip whitespace, lowercase
        return [str(item).strip().lower() for item in self.raw_data if item is not None]

    @functools.cached_property
    def token_pattern(self):
        """Compile a regex pattern once and cache it."""
        return re.compile(r'\w+')

    def search(self, term):
        """Use cached pattern to search in cleaned data."""
        return [item for item in self.cleaned_data if self.token_pattern.search(term)]

pipeline = DataPipeline(['Alice', None, 'Bob', 'CHARLIE', '  dave  '])

# First access triggers cleaning
print("First access of cleaned_data:")
result = pipeline.cleaned_data
print(f"  Result: {result}\n")

# Second access uses cache (no "CLEANING" output)
print("Second access of cleaned_data:")
result2 = pipeline.cleaned_data
print(f"  Result: {result2}\n")

# The pattern is also cached
print("Searching for 'alice'...")
print(f"  Found: {pipeline.search('alice')}")

# cached_property can be reset by deleting the attribute
print("\nDeleting cached_property to force recompute...")
del pipeline.cleaned_data
print("After delete, accessing again:")
result3 = pipeline.cleaned_data
print(f"  Result: {result3}")