Project 01 - Custom Decorator Library

Estimated time: 4–6 hours core | Level: Intermediate

Before reading the requirements, consider this question: a decorator is a function that takes a function and returns a function. A decorator factory is a function that takes arguments and returns a decorator. What does the call stack look like when you write @retry(max_attempts=3)? Trace it mentally before you write a line of code.

Learning Objectives

By the time this project is complete, you will have practiced:

• Writing decorator factories (decorators that take arguments)
• Using functools.wraps to preserve function metadata through decoration
• Using inspect.signature to introspect function arguments at call time
• Implementing state inside decorators using closures (rate limiter state, retry counters)
• Composing multiple decorators on a single function without breakage
• Exposing __wrapped__ so decorated functions can be introspected or unwrapped
• Understanding where each pattern is used in production Python frameworks

System Overview

You are building a standalone decorator library: a single module decorator_library.py that exports five decorators. Each decorator solves a real engineering problem that appears constantly in production Python code. The module has no external dependencies - standard library only.

Requirements

R1 - @retry(max_attempts=3, delay=1.0, exceptions=(Exception,))

Retries the decorated function up to max_attempts times if it raises any exception in exceptions. Between each retry, waits delay seconds, doubling the delay on each subsequent attempt (exponential backoff). After all attempts are exhausted, re-raises the last exception.

• max_attempts: total number of attempts including the first (not retries - attempts).
• delay: initial wait in seconds before the first retry.
• exceptions: a tuple of exception types to catch. Exceptions not in this tuple propagate immediately without retry.
• Each retry attempt must print (or log): Retry 2/3 after ValueError: bad input (attempt number, max, exception type, message).
• Must use functools.wraps.
• Must set __wrapped__ on the wrapper to point at the original function.

R2 - @rate_limit(calls=10, period=60)

Limits the decorated function to at most calls invocations per period seconds using a token bucket algorithm. If the limit is exceeded, raises RateLimitExceeded (a custom exception you define) with a message indicating how many seconds until the next call is allowed.

• State (call timestamps) lives in the closure, not in a global.
• The token bucket refills continuously: a call is allowed if fewer than calls timestamps in the last period seconds exist in the bucket.
• Thread safety is not required for the core requirement (extension challenge).
• Must use functools.wraps.
• Must set __wrapped__.

R3 - @timeout(seconds=5)

Raises TimeoutError if the decorated function takes longer than seconds to complete. Implement using threading.Timer or threading.Thread with a join timeout (not signal.alarm, which is UNIX-only and not safe in multithreaded programs).

• The function must run in a thread. If it completes within the timeout, return its result normally. If not, raise TimeoutError(f"Function exceeded {seconds}s timeout").
• Exception propagation: if the function raises an exception within the timeout, that exception must propagate to the caller - not TimeoutError.
• Must use functools.wraps.
• Must set __wrapped__.

R4 - @validate(schema)

Validates function arguments against a schema at call time. The schema is a dict mapping parameter names to types (e.g., {"name": str, "age": int}). If an argument's type does not match, raises TypeError with a descriptive message: TypeError: argument 'age' expected int, got str.

• Use inspect.signature to map positional arguments to parameter names - do not assume keyword-only calls.
• Only validate parameters listed in schema - unlisted parameters pass through unchecked.
• Must use functools.wraps.
• Must set __wrapped__.

R5 - @log_calls(logger=None, level=logging.DEBUG)

Logs every call to the decorated function: the function name, arguments, return value, and execution time in milliseconds. Uses the provided logger (a logging.Logger instance); if None, creates one named after the function using logging.getLogger.

• Log format on call entry: Calling greet(name='Alice', greeting='Hello')
• Log format on call exit: greet returned 'Hello, Alice!' in 2.3ms
• If the function raises, log: greet raised ValueError('bad input') after 1.1ms then re-raise.
• Must use functools.wraps.
• Must set __wrapped__.

R6 - Composability requirement

All five decorators must work correctly when stacked:

@log_calls()
@retry(max_attempts=3, delay=0.1)
@validate({"url": str, "timeout": int})
def fetch(url: str, timeout: int = 30) -> str:
    ...

The outermost decorator sees the wrapper from the next decorator, not the raw function. functools.wraps ensures that __name__, __doc__, and __wrapped__ are preserved through every layer. A function decorated with all five decorators must still report the correct __name__ and must have __wrapped__ pointing at the original.

R7 - .unwrap() utility

Implement a module-level unwrap(fn) function that follows the __wrapped__ chain and returns the original, undecorated function. This is useful for testing (call the raw function directly) and for introspection.

original = unwrap(fetch)
assert original.__name__ == "fetch"
assert original is not fetch

Starter Code Skeleton

import functools
import inspect
import logging
import threading
import time
from typing import Any, Callable, Tuple, Type


# ── Custom Exceptions ─────────────────────────────────────────────────────────

class RateLimitExceeded(Exception):
    pass


# ── Utilities ─────────────────────────────────────────────────────────────────

def unwrap(fn: Callable) -> Callable:
    """Follow __wrapped__ chain and return the original function."""
    # TODO: while fn has __wrapped__, reassign fn = fn.__wrapped__
    # TODO: return fn
    pass


# ── Retry ─────────────────────────────────────────────────────────────────────

def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
    """Retry decorator with exponential backoff."""
    def decorator(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # TODO: implement retry loop with exponential backoff
            # TODO: print retry message on each retry attempt
            # TODO: re-raise last exception after all attempts exhausted
            pass
        wrapper.__wrapped__ = fn
        return wrapper
    return decorator


# ── Rate Limit ────────────────────────────────────────────────────────────────

def rate_limit(calls: int = 10, period: float = 60.0):
    """Token bucket rate limiter decorator."""
    def decorator(fn: Callable) -> Callable:
        call_times: list[float] = []  # closure state

        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # TODO: remove timestamps older than `period` seconds from call_times
            # TODO: if len(call_times) >= calls, raise RateLimitExceeded
            # TODO: append current time to call_times
            # TODO: call and return fn(*args, **kwargs)
            pass
        wrapper.__wrapped__ = fn
        return wrapper
    return decorator


# ── Timeout ───────────────────────────────────────────────────────────────────

def timeout(seconds: float = 5.0):
    """Timeout decorator using threading."""
    def decorator(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            result: list[Any] = []
            exception: list[BaseException] = []

            def target():
                # TODO: call fn(*args, **kwargs)
                # TODO: on success, append result to result list
                # TODO: on exception, append exception to exception list
                pass

            thread = threading.Thread(target=target, daemon=True)
            # TODO: start thread
            # TODO: thread.join(timeout=seconds)
            # TODO: if thread is still alive, raise TimeoutError
            # TODO: if exception list is non-empty, re-raise
            # TODO: return result[0]
            pass
        wrapper.__wrapped__ = fn
        return wrapper
    return decorator


# ── Validate ──────────────────────────────────────────────────────────────────

def validate(schema: dict[str, type]):
    """Argument type validation decorator using inspect.signature."""
    def decorator(fn: Callable) -> Callable:
        sig = inspect.signature(fn)

        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # TODO: bind args and kwargs to sig to get a mapping of name -> value
            # TODO: apply defaults with bound.apply_defaults()
            # TODO: for each name in schema, check isinstance(bound.arguments[name], schema[name])
            # TODO: raise TypeError with descriptive message if type mismatch
            # TODO: call and return fn(*args, **kwargs)
            pass
        wrapper.__wrapped__ = fn
        return wrapper
    return decorator


# ── Log Calls ─────────────────────────────────────────────────────────────────

def log_calls(logger: logging.Logger = None, level: int = logging.DEBUG):
    """Structured call logging decorator with timing."""
    def decorator(fn: Callable) -> Callable:
        _logger = logger or logging.getLogger(fn.__name__)

        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # TODO: format args and kwargs into a readable call signature string
            # TODO: log "Calling fn_name(arg=val, ...)" at `level`
            # TODO: record start time
            # TODO: call fn(*args, **kwargs)
            # TODO: on success: log "fn_name returned <value> in Xms"
            # TODO: on exception: log "fn_name raised ExcType('msg') after Xms", re-raise
            pass
        wrapper.__wrapped__ = fn
        return wrapper
    return decorator


# ── Demonstration ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG, format="%(levelname)s %(name)s: %(message)s")

    # --- retry demo ---
    attempt_count = 0

    @retry(max_attempts=3, delay=0.1, exceptions=(ValueError,))
    def flaky_operation(x: int) -> int:
        global attempt_count
        attempt_count += 1
        if attempt_count < 3:
            raise ValueError(f"not ready yet (attempt {attempt_count})")
        return x * 2

    print("=== retry ===")
    print(flaky_operation(21))

    # --- rate_limit demo ---
    @rate_limit(calls=3, period=5.0)
    def api_call(endpoint: str) -> str:
        return f"OK: {endpoint}"

    print("\n=== rate_limit ===")
    for i in range(4):
        try:
            print(api_call(f"/resource/{i}"))
        except RateLimitExceeded as e:
            print(f"Rate limited: {e}")

    # --- timeout demo ---
    import time as _time

    @timeout(seconds=1.0)
    def slow_function(duration: float) -> str:
        _time.sleep(duration)
        return "done"

    print("\n=== timeout ===")
    print(slow_function(0.1))
    try:
        slow_function(2.0)
    except TimeoutError as e:
        print(f"Timed out: {e}")

    # --- validate demo ---
    @validate({"name": str, "age": int})
    def greet_user(name: str, age: int) -> str:
        return f"Hello {name}, age {age}"

    print("\n=== validate ===")
    print(greet_user("Alice", 30))
    try:
        greet_user("Bob", "thirty")
    except TypeError as e:
        print(f"Validation error: {e}")

    # --- log_calls demo ---
    @log_calls()
    def add(a: int, b: int) -> int:
        return a + b

    print("\n=== log_calls ===")
    add(3, 4)

    # --- composability demo ---
    @log_calls()
    @retry(max_attempts=2, delay=0.05, exceptions=(RuntimeError,))
    @validate({"value": int})
    def process(value: int) -> str:
        return f"processed: {value}"

    print("\n=== composed decorators ===")
    print(process(42))
    print(f"Function name preserved: {process.__name__}")
    print(f"Unwrapped: {unwrap(process).__name__}")

Expected Output

=== retry ===
Retry 2/3 after ValueError: not ready yet (attempt 1)
Retry 3/3 after ValueError: not ready yet (attempt 2)
42

=== rate_limit ===
OK: /resource/0
OK: /resource/1
OK: /resource/2
Rate limited: rate limit exceeded: 3 calls per 5.0s. Retry in X.Xs.

=== timeout ===
done
Timed out: Function exceeded 1.0s timeout

=== validate ===
Hello Alice, age 30
Validation error: argument 'age' expected int, got str

=== log_calls ===
DEBUG add: Calling add(a=3, b=4)
DEBUG add: add returned 7 in X.Xms

=== composed decorators ===
DEBUG process: Calling process(value=42)
DEBUG process: process returned 'processed: 42' in X.Xms
processed: 42
Function name preserved: process
Unwrapped: process

Note: timing values (X.Xms) and retry-in times will vary. Everything else must match exactly.

Step-by-Step Hints

Hint 1 - Start with unwrap and log_calls. log_calls has no retry logic, no threading, and no state - it is the cleanest decorator to build first. Once it works, you have a working template for all the others. unwrap is five lines. Write it first.

Hint 2 - Decorator factories have three levels of nesting. retry(max_attempts=3) returns decorator. decorator(fn) returns wrapper. wrapper(*args, **kwargs) is what actually runs. If you lose track of which level you are at, draw it out:

retry(max_attempts=3)         # level 1: factory - returns decorator
  └─ decorator(fn)            # level 2: decorator - returns wrapper
       └─ wrapper(*args, **kwargs)  # level 3: wrapper - runs on every call

Hint 3 - functools.wraps copies __name__, __doc__, __annotations__, and __wrapped__ automatically. But __wrapped__ is set to point at fn only if you set it explicitly after applying @functools.wraps(fn). The functools.wraps decorator sets __wrapped__ automatically when you call it - but set it explicitly anyway for clarity and to make unwrap work correctly through all layers.

Hint 4 - inspect.signature for @validate.

sig = inspect.signature(fn)           # bind once at decoration time, not at call time
bound = sig.bind(*args, **kwargs)     # at call time
bound.apply_defaults()                # fills in default values
# bound.arguments is now an OrderedDict of name -> value

Bind at decoration time (outside wrapper) to pay the introspection cost once, not on every call.

Hint 5 - Thread-based timeout result passing. A thread function cannot return a value to the caller. Use mutable containers to communicate:

result = []
exception = []

def target():
    try:
        result.append(fn(*args, **kwargs))
    except Exception as e:
        exception.append(e)

After thread.join(timeout=seconds), check thread.is_alive() first (timeout exceeded), then exception (function raised), then result[0] (success).

Hint 6 - Rate limiter uses a sliding window.

now = time.monotonic()
# Remove timestamps outside the window
call_times[:] = [t for t in call_times if now - t < period]
if len(call_times) >= calls:
    oldest = call_times[0]
    retry_after = period - (now - oldest)
    raise RateLimitExceeded(f"... Retry in {retry_after:.1f}s.")
call_times.append(now)

call_times[:] = [...] mutates the list in-place, which matters because the list is shared via closure.

Hint 7 - Exponential backoff in @retry.

current_delay = delay
for attempt in range(1, max_attempts + 1):
    try:
        return fn(*args, **kwargs)
    except exceptions as e:
        if attempt == max_attempts:
            raise
        print(f"Retry {attempt + 1}/{max_attempts} after {type(e).__name__}: {e}")
        time.sleep(current_delay)
        current_delay *= 2

FP Concepts Tested

Concept	Where it appears
Closures	All decorators - state (call_times, attempt_count) lives in closure
Higher-order functions	Every decorator takes and returns a function
Decorator factories	All five - each takes arguments and returns a decorator
functools.wraps	Applied in every decorator's inner wrapper
inspect.signature	@validate - binds positional args to parameter names
Threading for isolation	@timeout - function runs in a thread with a join timeout
Immutability via __wrapped__	unwrap() follows the chain; original is never modified

Engineering Notes - Where These Patterns Appear in Production

@retry is the backbone of resilience in distributed systems. Celery uses retry logic with exponential backoff for task queues. The tenacity library is a production-grade retry library built on exactly this pattern. AWS SDK (boto3) implements automatic retries with jitter for API calls.

@rate_limit appears in API client libraries everywhere - any library that wraps a third-party API (Twitter, Stripe, GitHub) implements rate limiting to stay within API quotas. FastAPI uses slowapi (which wraps limits) for route-level rate limiting. The sliding window token bucket approach you implement here is the standard algorithm.

@timeout is used in web frameworks to prevent slow database queries or external HTTP calls from blocking threads indefinitely. Django's CONN_MAX_AGE for database connections and socket.setdefaulttimeout() are related mechanisms. The thread-based approach you implement is the safe alternative to signal.alarm, which cannot be used in web server worker threads.

@validate is the conceptual predecessor to Pydantic. FastAPI's parameter validation, before Pydantic takes over, uses inspect.signature to map HTTP request parameters to function arguments. The pattern of binding at decoration time (not call time) is a performance optimization that FastAPI applies.

@log_calls is standard in any production observability stack. OpenTelemetry's Python SDK instruments functions with decorators that capture timing and argument metadata. The Django ORM's query logging uses a similar wrapper pattern. Structured logging (emitting key-value pairs rather than formatted strings) is what separates production logging from debug prints.

Extension Challenges

Extension 1 - Thread-safe rate limiter Add a threading.Lock to @rate_limit so it is safe when multiple threads call the same decorated function concurrently. Without the lock, two threads can both read call_times, both find it under the limit, both append, and collectively exceed the limit.

Extension 2 - @cache(maxsize=128) Implement functools.lru_cache from scratch as a decorator factory. The cache stores results keyed by (args, frozenset(kwargs.items())). Implement cache_info() (hits, misses, current size) and cache_clear() as attributes on the wrapper.

Extension 3 - Decorator introspection Add a decorators(fn) utility that returns a list of decorator names applied to a function, by following the __wrapped__ chain and inspecting each wrapper's __name__. Stack-trace-style output: ['log_calls', 'retry', 'validate', 'fetch'].

Extension 4 - Async decorator support Modify all five decorators to work with both sync and async functions. Use inspect.iscoroutinefunction(fn) to detect async functions and branch accordingly. The @timeout decorator for async functions should use asyncio.wait_for rather than threading.

Extension 5 - @circuit_breaker(failure_threshold=5, recovery_timeout=30) Implement the circuit breaker pattern: after failure_threshold consecutive failures, the decorator stops calling the function and immediately raises CircuitOpenError for recovery_timeout seconds, then allows one probe call through (half-open state). This is the pattern used in microservice resilience libraries like pybreaker.