Project: Production Decorators (Retry, Cache, and Rate-Limiter)0%

Project: Production Decorators (Retry, Cache, and Rate-Limiter)

Intermediate16 min read•Updated: 2026-10-01

Project: Production Decorators (Retry, Cache, and Rate-Limiter)

Welcome to the Chapter 2 Capstone Project! In modern microservices and backend web engineering, decorators provide non-invasive resilience, security, and performance optimizations.

In this project, you will build a production-grade UPI Banking Resilience Decorator Suite incorporating three industry-standard patterns:

1
Decorators with Arguments: The three-tier closure pattern (def decorator_factory(param): def decorator(func): def wrapper(...)).
2
Exponential Retry Decorator (@retry_on_failure): Automatically retries transient network or database errors with exponential backoff.
3
Memoization Cache Decorator (@memoize): Eliminates redundant compute and database queries by caching return values in memory.
4
Sliding-Window Rate Limiter (@rate_limit): Protects sensitive financial APIs against high-frequency abuse.

Real-World Analogy: The UPI Payment Gateway Shield

Imagine the payment processing pipeline powering Indian UPI transactions (PhonePe, Google Pay, Paytm):

Visual Architecture Blueprint

+-------------------------------------------------------------------------+
|                  UPI PAYMENT GATEWAY RESILIENCE SHIELD                  |
+-------------------------------------------------------------------------+
|                                                                         |
|  Incoming Customer Payment: UPI ₹500 via HDFC Bank                     |
|                               │                                         |
|                               ▼                                         |
|  1. @rate_limit(max_per_sec=3) ──> Blocks automated bots & spamming     |
|                               │ (Allowed)                               |
|                               ▼                                         |
|  2. @memoize_cache             ──> Instantly returns cached bank IFSC &  |
|                                    branch coordinates without DB hit    |
|                               │ (Not cached yet)                        |
|                               ▼                                         |
|  3. @retry_on_failure(3)      ──> If cellular tower drops packet,       |
|                                    silently retries bank switch 3 times |
|                               │ (Success!)                              |
|                               ▼                                         |
|  Core Banking Service: process_payment() Executes Successfully!         |
|                                                                         |
+-------------------------------------------------------------------------+

By wrapping our core payment function with these modular decorators, the business logic remains clean, readable, and 100% focused on financial accounting, while the decorators handle reliability and defense.

Three-Tier Architecture: Decorators That Accept Arguments

When a decorator needs configuration parameters (e.g. @retry(max_attempts=3)), it requires three levels of nested functions:

Visual Architecture Blueprint

+-------------------------------------------------------------------------+
|                 THREE-TIER PARAMETERIZED DECORATOR PATTERN              |
+-------------------------------------------------------------------------+
|                                                                         |
|  def retry(max_attempts=3):          <-- 1. Decorator Factory (takes cfg)|
|      def actual_decorator(func):     <-- 2. Decorator (takes function)  |
|          @wraps(func)                                                   |
|          def wrapper(*args, **kwargs):<-- 3. Wrapper Closure (at call)   |
|              # Execution logic using max_attempts and func              |
|              return func(*args, **kwargs)                               |
|          return wrapper                                                 |
|      return actual_decorator                                            |
|                                                                         |
+-------------------------------------------------------------------------+

Complete Production-Grade Implementation

Here is the complete, modular, runnable code for the Decorator Suite:

1"""
2MSK Python Capstone: Industrial Resilience Decorator Suite
3Author: MSK Institute
4"""
5import time
6from functools import wraps
7 
8# -------------------------------------------------------------
9# Decorator 1: In-Memory Memoization Cache
10# -------------------------------------------------------------
11def memoize(func):
12    """Caches deterministic function results using an in-memory dictionary."""
13    cache = {}
14 
15    @wraps(func)
16    def wrapper(*args):
17        # Arguments serve as composite hash key
18        if args in cache:
19            print(f"[CACHE HIT] Returning cached result for {func.__name__}{args}")
20            return cache[args]
21        
22        print(f"[CACHE MISS] Computing new result for {func.__name__}{args}")
23        result = func(*args)
24        cache[args] = result
25        return result
26 
27    return wrapper
28 
29# -------------------------------------------------------------
30# Decorator 2: Parameterized Retry on Failure
31# -------------------------------------------------------------
32def retry_on_failure(max_retries=3, delay_seconds=0.1, allowed_exceptions=(Exception,)):
33    """Retries a failing function up to max_retries times before propagating error."""
34    def decorator(func):
35        @wraps(func)
36        def wrapper(*args, **kwargs):
37            attempts = 0
38            while attempts < max_retries:
39                try:
40                    return func(*args, **kwargs)
41                except allowed_exceptions as err:
42                    attempts += 1
43                    print(f"[RETRY WARNING] Attempt {attempts}/{max_retries} failed for '{func.__name__}': {err}")
44                    if attempts >= max_retries:
45                        print(f"[RETRY EXHAUSTED] Max retries reached. Raising exception.")
46                        raise err
47                    time.sleep(delay_seconds)
48        return wrapper
49    return decorator
50 
51# -------------------------------------------------------------
52# Decorator 3: Sliding-Window Rate Limiter
53# -------------------------------------------------------------
54def rate_limiter(max_calls=3, period_seconds=1.0):
55    """Limits the number of function executions within a specified time window."""
56    def decorator(func):
57        invocation_timestamps = []
58 
59        @wraps(func)
60        def wrapper(*args, **kwargs):
61            nonlocal invocation_timestamps
62            current_time = time.time()
63            
64            # Prune timestamps older than the sliding window
65            invocation_timestamps = [t for t in invocation_timestamps if current_time - t <= period_seconds]
66            
67            if len(invocation_timestamps) >= max_calls:
68                raise PermissionError(
69                    f"Rate limit exceeded for '{func.__name__}'! Allowed: {max_calls} calls per {period_seconds}s."
70                )
71            
72            invocation_timestamps.append(current_time)
73            return func(*args, **kwargs)
74        return wrapper
75    return decorator
76 
77 
78# -------------------------------------------------------------
79# Demonstration & Verification
80# -------------------------------------------------------------
81 
82# Test Service 1: IFSC Branch Lookup (Demonstrating Memoization)
83@memoize
84def fetch_bank_branch(ifsc_code):
85    """Simulates a heavy remote database lookup for bank IFSC codes."""
86    time.sleep(0.05)  # Simulate network latency
87    mock_db = {
88        "SBIN000101": {"bank": "State Bank of India", "branch": "Connaught Place, New Delhi"},
89        "HDFC000452": {"bank": "HDFC Bank", "branch": "Bandra West, Mumbai"}
90    }
91    return mock_db.get(ifsc_code, {"error": "Invalid IFSC"})
92 
93# Test Service 2: Unstable UPI Server Connection (Demonstrating Retry)
94flaky_counter = 0
95 
96@retry_on_failure(max_retries=3, delay_seconds=0.05, allowed_exceptions=(ConnectionError,))
97def connect_to_npci_switch(vpa_id):
98    """Simulates a network call that fails twice before succeeding."""
99    global flaky_counter
100    flaky_counter += 1
101    if flaky_counter < 3:
102        raise ConnectionError("NPCI Central Switch timeout: Packet dropped on cellular link")
103    return f"CONNECTED to NPCI for VPA '{vpa_id}' on attempt {flaky_counter}"
104 
105# Test Service 3: Fast Withdrawal API (Demonstrating Rate Limiting)
106@rate_limiter(max_calls=2, period_seconds=0.5)
107def send_otp(phone_number):
108    return f"OTP sent to +91 {phone_number}"
109 
110 
111if __name__ == "__main__":
112    print("=== 1. Testing Memoization Cache Decorator ===")
113    print(fetch_bank_branch("SBIN000101"))
114    # Second call should hit the cache immediately!
115    print(fetch_bank_branch("SBIN000101"))
116 
117    print("\n=== 2. Testing Retry Mechanism Decorator ===")
118    connection_status = connect_to_npci_switch("user@oksbi")
119    print(f"Final Status: {connection_status}")
120 
121    print("\n=== 3. Testing Sliding-Window Rate Limiter ===")
122    print(send_otp("9876543210"))
123    print(send_otp("9876543210"))
124    try:
125        # Third call within 0.5s window must be blocked
126        print(send_otp("9876543210"))
127    except PermissionError as e:
128        print(f"[BLOCKED AS EXPECTED] {e}")

Expected Output

1=== 1. Testing Memoization Cache Decorator ===
2[CACHE MISS] Computing new result for fetch_bank_branch('SBIN000101',)
3{'bank': 'State Bank of India', 'branch': 'Connaught Place, New Delhi'}
4[CACHE HIT] Returning cached result for fetch_bank_branch('SBIN000101',)
5{'bank': 'State Bank of India', 'branch': 'Connaught Place, New Delhi'}
6 
7=== 2. Testing Retry Mechanism Decorator ===
8[RETRY WARNING] Attempt 1/3 failed for 'connect_to_npci_switch': NPCI Central Switch timeout: Packet dropped on cellular link
9[RETRY WARNING] Attempt 2/3 failed for 'connect_to_npci_switch': NPCI Central Switch timeout: Packet dropped on cellular link
10Final Status: CONNECTED to NPCI for VPA 'user@oksbi' on attempt 3
11 
12=== 3. Testing Sliding-Window Rate Limiter ===
13OTP sent to +91 9876543210
14OTP sent to +91 9876543210
15[BLOCKED AS EXPECTED] Rate limit exceeded for 'send_otp'! Allowed: 2 calls per 0.5s.

Best Practices & Comparison: Do's and Don'ts

Practice	Bad Implementation	Gold-Standard Implementation
Parameterized Decorators	Trying to pass arguments to a 2-tier decorator	Use 3-tier pattern: Factory $\to$ Decorator $\to$ Wrapper
Retry Interception	Catching all exceptions (`except:`) blindly	Catch only transient network errors (`allowed_exceptions=(...)`)
Cache Keys	Using unhashable objects as cache keys	Normalize arguments to tuples or immutable representations
Cache Growth	Allowing unbounded cache memory growth	Use LRU (Least Recently Used) cache or bounded dictionary size
Rate Limiting	Static counters that never reset	Sliding-window timestamp pruning with `time.time()`

Quick Revision Summary Cheat Sheet

Parameterized Decorator: def factory(config): def dec(func): def wrapper(*args, **kw): ...
Memoization: Storing cache[args] = func(*args) avoids expensive re-computation of pure functions.
Resilience: @retry_on_failure turns fragile network calls into fault-tolerant distributed operations.
Security: @rate_limiter uses timestamp filtering to prevent denial-of-service and brute-force attacks.
Composition: Multiple decorators can be stacked together to create resilient, cached, and secure endpoints.

Practice Challenge

Scenario: Structured Execution Logger Decorator with File Output

Build a logging decorator @log_to_file(filepath="app_audit.log") that:

1
Accepts an optional filepath argument (defaulting to "app_audit.log").
2
Intercepts function execution and records:

Current ISO timestamp.
Function name.
Arguments passed (args and kwargs).
Function return value or exception raised.

1
Appends the log record to the specified file using the with open(filepath, "a") context manager.
2
Preserves function metadata using @wraps.

Starter Code

1from functools import wraps
2from datetime import datetime
3 
4def log_to_file(filepath="app_audit.log"):
5    # TODO: Implement 3-tier parameterized decorator with file appending
6    pass

Complete Solution

1import os
2from functools import wraps
3from datetime import datetime
4 
5def log_to_file(filepath="app_audit.log"):
6    def decorator(func):
7        @wraps(func)
8        def wrapper(*args, **kwargs):
9            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
10            try:
11                result = func(*args, **kwargs)
12                log_entry = f"[{timestamp}] SUCCESS: {func.__name__} args={args} kwargs={kwargs} -> {result}\n"
13                with open(filepath, "a", encoding="utf-8") as f:
14                    f.write(log_entry)
15                return result
16            except Exception as err:
17                log_entry = f"[{timestamp}] ERROR: {func.__name__} args={args} kwargs={kwargs} -> {type(err).__name__}: {err}\n"
18                with open(filepath, "a", encoding="utf-8") as f:
19                    f.write(log_entry)
20                raise err
21        return wrapper
22    return decorator
23 
24# Test Service
25AUDIT_LOG = "test_audit.log"
26if os.path.exists(AUDIT_LOG):
27    os.remove(AUDIT_LOG)
28 
29@log_to_file(filepath=AUDIT_LOG)
30def transfer_upi(sender, receiver, amount):
31    if amount <= 0:
32        raise ValueError("Transfer amount must be positive!")
33    return f"TXN-{int(amount * 100)}"
34 
35# Run valid transaction
36t1 = transfer_upi("arjun@sbi", "priya@hdfc", 1500.0)
37print("Transfer 1:", t1)
38 
39# Run invalid transaction
40try:
41    transfer_upi("arjun@sbi", "priya@hdfc", -50.0)
42except ValueError:
43    print("Caught expected ValueError on negative transfer.")
44 
45# Verify log file content
46print("\n--- Generated Audit Log File ---")
47with open(AUDIT_LOG, "r", encoding="utf-8") as f:
48    print(f.read().strip())

Expected Output

1Transfer 1: TXN-150000
2Caught expected ValueError on negative transfer.
3 
4--- Generated Audit Log File ---
5[2026-09-12 15:05:00] SUCCESS: transfer_upi args=('arjun@sbi', 'priya@hdfc', 1500.0) kwargs={} -> TXN-150000
6[2026-09-12 15:05:00] ERROR: transfer_upi args=('arjun@sbi', 'priya@hdfc', -50.0) kwargs={} -> ValueError: Transfer amount must be positive!

Interactive Quiz• Knowledge Check

Practice Quiz

Question 1

How many levels of nested functions are required to implement a decorator that accepts configuration arguments, like @retry(max_retries=5)?

Question 2

In the memoization decorator, why can args be used directly as a dictionary cache key?

Question 3

What is the danger of writing an unconstrained retry decorator with while True and no maximum attempt limit?

Question 4

How does the sliding-window rate limiter prune expired request timestamps?

Question 5

What will happen if you apply both @memoize and @retry_on_failure to a function?

🚀 What's Next?

In the next lesson, we will continue your intermediate Python journey with Creating and Importing Modules (3: Modules and Packages).

👉 Continue to Next Lesson: Creating and Importing Modules →

Next Lesson

Creating and Importing Modules

Continue learning with hands-on practice, examples, and exercises in the upcoming topic.

Related Lessons

Previous Lesson	Next Lesson
Decorators Basics	Creating and Importing Modules

Practice Quiz

Test your understanding of this lesson with 5 questions. Each question has one correct answer.

Previous Next

PreviousPrevNext

MSK Institute • Shikohabad

Loading Page Content...

Preparing courses, live labs, and learning resources.सामग्री लोड हो रही है, कृपया प्रतीक्षा करें...

Connecting serverPlease wait...

💻 Practical Labs🚀 Free Demo📜 Verified Certificates

Creating and Importing Modules

Project: Production Decorators (Retry, Cache, and Rate-Limiter)

Intermediate16 min read•Updated: 2026-10-01

Study Materials Cheatsheet

Project: Production Decorators (Retry, Cache, and Rate-Limiter)

Welcome to the Chapter 2 Capstone Project! In modern microservices and backend web engineering, decorators provide non-invasive resilience, security, and performance optimizations.

In this project, you will build a production-grade UPI Banking Resilience Decorator Suite incorporating three industry-standard patterns:

1
Decorators with Arguments: The three-tier closure pattern (def decorator_factory(param): def decorator(func): def wrapper(...)).
2
Exponential Retry Decorator (@retry_on_failure): Automatically retries transient network or database errors with exponential backoff.
3
Memoization Cache Decorator (@memoize): Eliminates redundant compute and database queries by caching return values in memory.
4
Sliding-Window Rate Limiter (@rate_limit): Protects sensitive financial APIs against high-frequency abuse.

Real-World Analogy: The UPI Payment Gateway Shield

Imagine the payment processing pipeline powering Indian UPI transactions (PhonePe, Google Pay, Paytm):

Visual Architecture Blueprint

+-------------------------------------------------------------------------+
|                  UPI PAYMENT GATEWAY RESILIENCE SHIELD                  |
+-------------------------------------------------------------------------+
|                                                                         |
|  Incoming Customer Payment: UPI ₹500 via HDFC Bank                     |
|                               │                                         |
|                               ▼                                         |
|  1. @rate_limit(max_per_sec=3) ──> Blocks automated bots & spamming     |
|                               │ (Allowed)                               |
|                               ▼                                         |
|  2. @memoize_cache             ──> Instantly returns cached bank IFSC &  |
|                                    branch coordinates without DB hit    |
|                               │ (Not cached yet)                        |
|                               ▼                                         |
|  3. @retry_on_failure(3)      ──> If cellular tower drops packet,       |
|                                    silently retries bank switch 3 times |
|                               │ (Success!)                              |
|                               ▼                                         |
|  Core Banking Service: process_payment() Executes Successfully!         |
|                                                                         |
+-------------------------------------------------------------------------+

Three-Tier Architecture: Decorators That Accept Arguments

When a decorator needs configuration parameters (e.g. @retry(max_attempts=3)), it requires three levels of nested functions:

Visual Architecture Blueprint

+-------------------------------------------------------------------------+
|                 THREE-TIER PARAMETERIZED DECORATOR PATTERN              |
+-------------------------------------------------------------------------+
|                                                                         |
|  def retry(max_attempts=3):          <-- 1. Decorator Factory (takes cfg)|
|      def actual_decorator(func):     <-- 2. Decorator (takes function)  |
|          @wraps(func)                                                   |
|          def wrapper(*args, **kwargs):<-- 3. Wrapper Closure (at call)   |
|              # Execution logic using max_attempts and func              |
|              return func(*args, **kwargs)                               |
|          return wrapper                                                 |
|      return actual_decorator                                            |
|                                                                         |
+-------------------------------------------------------------------------+

Complete Production-Grade Implementation

Here is the complete, modular, runnable code for the Decorator Suite:

1"""
2MSK Python Capstone: Industrial Resilience Decorator Suite
3Author: MSK Institute
4"""
5import time
6from functools import wraps
7 
8# -------------------------------------------------------------
9# Decorator 1: In-Memory Memoization Cache
10# -------------------------------------------------------------
11def memoize(func):
12    """Caches deterministic function results using an in-memory dictionary."""
13    cache = {}
14 
15    @wraps(func)
16    def wrapper(*args):
17        # Arguments serve as composite hash key
18        if args in cache:
19            print(f"[CACHE HIT] Returning cached result for {func.__name__}{args}")
20            return cache[args]
21        
22        print(f"[CACHE MISS] Computing new result for {func.__name__}{args}")
23        result = func(*args)
24        cache[args] = result
25        return result
26 
27    return wrapper
28 
29# -------------------------------------------------------------
30# Decorator 2: Parameterized Retry on Failure
31# -------------------------------------------------------------
32def retry_on_failure(max_retries=3, delay_seconds=0.1, allowed_exceptions=(Exception,)):
33    """Retries a failing function up to max_retries times before propagating error."""
34    def decorator(func):
35        @wraps(func)
36        def wrapper(*args, **kwargs):
37            attempts = 0
38            while attempts < max_retries:
39                try:
40                    return func(*args, **kwargs)
41                except allowed_exceptions as err:
42                    attempts += 1
43                    print(f"[RETRY WARNING] Attempt {attempts}/{max_retries} failed for '{func.__name__}': {err}")
44                    if attempts >= max_retries:
45                        print(f"[RETRY EXHAUSTED] Max retries reached. Raising exception.")
46                        raise err
47                    time.sleep(delay_seconds)
48        return wrapper
49    return decorator
50 
51# -------------------------------------------------------------
52# Decorator 3: Sliding-Window Rate Limiter
53# -------------------------------------------------------------
54def rate_limiter(max_calls=3, period_seconds=1.0):
55    """Limits the number of function executions within a specified time window."""
56    def decorator(func):
57        invocation_timestamps = []
58 
59        @wraps(func)
60        def wrapper(*args, **kwargs):
61            nonlocal invocation_timestamps
62            current_time = time.time()
63            
64            # Prune timestamps older than the sliding window
65            invocation_timestamps = [t for t in invocation_timestamps if current_time - t <= period_seconds]
66            
67            if len(invocation_timestamps) >= max_calls:
68                raise PermissionError(
69                    f"Rate limit exceeded for '{func.__name__}'! Allowed: {max_calls} calls per {period_seconds}s."
70                )
71            
72            invocation_timestamps.append(current_time)
73            return func(*args, **kwargs)
74        return wrapper
75    return decorator
76 
77 
78# -------------------------------------------------------------
79# Demonstration & Verification
80# -------------------------------------------------------------
81 
82# Test Service 1: IFSC Branch Lookup (Demonstrating Memoization)
83@memoize
84def fetch_bank_branch(ifsc_code):
85    """Simulates a heavy remote database lookup for bank IFSC codes."""
86    time.sleep(0.05)  # Simulate network latency
87    mock_db = {
88        "SBIN000101": {"bank": "State Bank of India", "branch": "Connaught Place, New Delhi"},
89        "HDFC000452": {"bank": "HDFC Bank", "branch": "Bandra West, Mumbai"}
90    }
91    return mock_db.get(ifsc_code, {"error": "Invalid IFSC"})
92 
93# Test Service 2: Unstable UPI Server Connection (Demonstrating Retry)
94flaky_counter = 0
95 
96@retry_on_failure(max_retries=3, delay_seconds=0.05, allowed_exceptions=(ConnectionError,))
97def connect_to_npci_switch(vpa_id):
98    """Simulates a network call that fails twice before succeeding."""
99    global flaky_counter
100    flaky_counter += 1
101    if flaky_counter < 3:
102        raise ConnectionError("NPCI Central Switch timeout: Packet dropped on cellular link")
103    return f"CONNECTED to NPCI for VPA '{vpa_id}' on attempt {flaky_counter}"
104 
105# Test Service 3: Fast Withdrawal API (Demonstrating Rate Limiting)
106@rate_limiter(max_calls=2, period_seconds=0.5)
107def send_otp(phone_number):
108    return f"OTP sent to +91 {phone_number}"
109 
110 
111if __name__ == "__main__":
112    print("=== 1. Testing Memoization Cache Decorator ===")
113    print(fetch_bank_branch("SBIN000101"))
114    # Second call should hit the cache immediately!
115    print(fetch_bank_branch("SBIN000101"))
116 
117    print("\n=== 2. Testing Retry Mechanism Decorator ===")
118    connection_status = connect_to_npci_switch("user@oksbi")
119    print(f"Final Status: {connection_status}")
120 
121    print("\n=== 3. Testing Sliding-Window Rate Limiter ===")
122    print(send_otp("9876543210"))
123    print(send_otp("9876543210"))
124    try:
125        # Third call within 0.5s window must be blocked
126        print(send_otp("9876543210"))
127    except PermissionError as e:
128        print(f"[BLOCKED AS EXPECTED] {e}")

Expected Output

1=== 1. Testing Memoization Cache Decorator ===
2[CACHE MISS] Computing new result for fetch_bank_branch('SBIN000101',)
3{'bank': 'State Bank of India', 'branch': 'Connaught Place, New Delhi'}
4[CACHE HIT] Returning cached result for fetch_bank_branch('SBIN000101',)
5{'bank': 'State Bank of India', 'branch': 'Connaught Place, New Delhi'}
6 
7=== 2. Testing Retry Mechanism Decorator ===
8[RETRY WARNING] Attempt 1/3 failed for 'connect_to_npci_switch': NPCI Central Switch timeout: Packet dropped on cellular link
9[RETRY WARNING] Attempt 2/3 failed for 'connect_to_npci_switch': NPCI Central Switch timeout: Packet dropped on cellular link
10Final Status: CONNECTED to NPCI for VPA 'user@oksbi' on attempt 3
11 
12=== 3. Testing Sliding-Window Rate Limiter ===
13OTP sent to +91 9876543210
14OTP sent to +91 9876543210
15[BLOCKED AS EXPECTED] Rate limit exceeded for 'send_otp'! Allowed: 2 calls per 0.5s.

Best Practices & Comparison: Do's and Don'ts

Practice	Bad Implementation	Gold-Standard Implementation
Parameterized Decorators	Trying to pass arguments to a 2-tier decorator	Use 3-tier pattern: Factory $\to$ Decorator $\to$ Wrapper
Retry Interception	Catching all exceptions (`except:`) blindly	Catch only transient network errors (`allowed_exceptions=(...)`)
Cache Keys	Using unhashable objects as cache keys	Normalize arguments to tuples or immutable representations
Cache Growth	Allowing unbounded cache memory growth	Use LRU (Least Recently Used) cache or bounded dictionary size
Rate Limiting	Static counters that never reset	Sliding-window timestamp pruning with `time.time()`

Quick Revision Summary Cheat Sheet

Parameterized Decorator: def factory(config): def dec(func): def wrapper(*args, **kw): ...
Memoization: Storing cache[args] = func(*args) avoids expensive re-computation of pure functions.
Resilience: @retry_on_failure turns fragile network calls into fault-tolerant distributed operations.
Security: @rate_limiter uses timestamp filtering to prevent denial-of-service and brute-force attacks.
Composition: Multiple decorators can be stacked together to create resilient, cached, and secure endpoints.

Practice Challenge

Scenario: Structured Execution Logger Decorator with File Output

Build a logging decorator @log_to_file(filepath="app_audit.log") that:

1
Accepts an optional filepath argument (defaulting to "app_audit.log").
2
Intercepts function execution and records:

Current ISO timestamp.
Function name.
Arguments passed (args and kwargs).
Function return value or exception raised.

1
Appends the log record to the specified file using the with open(filepath, "a") context manager.
2
Preserves function metadata using @wraps.

Starter Code

1from functools import wraps
2from datetime import datetime
3 
4def log_to_file(filepath="app_audit.log"):
5    # TODO: Implement 3-tier parameterized decorator with file appending
6    pass

Complete Solution

1import os
2from functools import wraps
3from datetime import datetime
4 
5def log_to_file(filepath="app_audit.log"):
6    def decorator(func):
7        @wraps(func)
8        def wrapper(*args, **kwargs):
9            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
10            try:
11                result = func(*args, **kwargs)
12                log_entry = f"[{timestamp}] SUCCESS: {func.__name__} args={args} kwargs={kwargs} -> {result}\n"
13                with open(filepath, "a", encoding="utf-8") as f:
14                    f.write(log_entry)
15                return result
16            except Exception as err:
17                log_entry = f"[{timestamp}] ERROR: {func.__name__} args={args} kwargs={kwargs} -> {type(err).__name__}: {err}\n"
18                with open(filepath, "a", encoding="utf-8") as f:
19                    f.write(log_entry)
20                raise err
21        return wrapper
22    return decorator
23 
24# Test Service
25AUDIT_LOG = "test_audit.log"
26if os.path.exists(AUDIT_LOG):
27    os.remove(AUDIT_LOG)
28 
29@log_to_file(filepath=AUDIT_LOG)
30def transfer_upi(sender, receiver, amount):
31    if amount <= 0:
32        raise ValueError("Transfer amount must be positive!")
33    return f"TXN-{int(amount * 100)}"
34 
35# Run valid transaction
36t1 = transfer_upi("arjun@sbi", "priya@hdfc", 1500.0)
37print("Transfer 1:", t1)
38 
39# Run invalid transaction
40try:
41    transfer_upi("arjun@sbi", "priya@hdfc", -50.0)
42except ValueError:
43    print("Caught expected ValueError on negative transfer.")
44 
45# Verify log file content
46print("\n--- Generated Audit Log File ---")
47with open(AUDIT_LOG, "r", encoding="utf-8") as f:
48    print(f.read().strip())

Expected Output

1Transfer 1: TXN-150000
2Caught expected ValueError on negative transfer.
3 
4--- Generated Audit Log File ---
5[2026-09-12 15:05:00] SUCCESS: transfer_upi args=('arjun@sbi', 'priya@hdfc', 1500.0) kwargs={} -> TXN-150000
6[2026-09-12 15:05:00] ERROR: transfer_upi args=('arjun@sbi', 'priya@hdfc', -50.0) kwargs={} -> ValueError: Transfer amount must be positive!

Interactive Quiz• Knowledge Check

Practice Quiz

Question 1

How many levels of nested functions are required to implement a decorator that accepts configuration arguments, like @retry(max_retries=5)?

Question 2

In the memoization decorator, why can args be used directly as a dictionary cache key?

Question 3

What is the danger of writing an unconstrained retry decorator with while True and no maximum attempt limit?

Question 4

How does the sliding-window rate limiter prune expired request timestamps?

Question 5

What will happen if you apply both @memoize and @retry_on_failure to a function?

🚀 What's Next?

In the next lesson, we will continue your intermediate Python journey with Creating and Importing Modules (3: Modules and Packages).

👉 Continue to Next Lesson: Creating and Importing Modules →

Next Lesson

Creating and Importing Modules

Continue learning with hands-on practice, examples, and exercises in the upcoming topic.

Related Lessons

Previous Lesson	Next Lesson
Decorators Basics	Creating and Importing Modules

Practice Quiz

Test your understanding of this lesson with 5 questions. Each question has one correct answer.

Previous Next