Advanced Topics: Decorators, Generators, Context Managers
Decorators are like gift wrappers for functions, Generators are like lazy factory workers producing items one at a time, and Context Managers are like responsible babysitters who clean up after themselves.
These three advanced Python features allow you to write more elegant, efficient, and maintainable code. **Decorators** modify the behaviour of functions or methods without changing their source code. **Generators** produce sequences of values lazily, saving memory. **Context managers** automatically handle setup and teardown (e.g., closing files). Mastering them will make your Python code more Pythonic and professional.
---
## 1. DECORATORS
A **decorator** is a function that takes another function as an argument, extends its behaviour, and returns the modified function. Decorators are often used for logging, timing, access control, caching, and validation.
**Basic syntax:**
```python
def my_decorator(func):
def wrapper():
# code before
func()
# code after
return wrapper
@my_decorator
def say_hello():
print("Hello")
```
The `@my_decorator` is syntactic sugar for `say_hello = my_decorator(say_hello)`.
**Decorators with arguments:**
If you need to pass parameters to the decorator itself (like `repeat(3)`), you need an extra level of nesting: a function that returns a decorator.
**Preserving metadata:**
The `@wraps(func)` from `functools` copies the original function's name, docstring, and annotations to the wrapper; otherwise, introspection tools would see the wrapper instead of the original.
**Common uses:**
- **Timing**: measure execution time (`@timer`).
- **Caching (memoization)**: store results of expensive function calls (`@cache`).
- **Authorization**: check user roles before executing (`@requires_auth`).
- **Logging**: automatically log arguments and return values.
---
## 2. GENERATORS
A **generator** is a function that uses `yield` instead of `return`. It produces a sequence of values lazily – one at a time, only when requested. This is memory‑efficient for large or infinite sequences because you don't store all values in memory at once.
**Basic generator:**
```python
def count_up_to(n):
i = 1
while i <= n:
yield i
i += 1
```
Each call to `next(generator)` resumes execution right after the `yield` until the next `yield` or the function ends. When exhausted, the generator raises `StopIteration`.
**Generator expressions:**
Similar to list comprehensions but with parentheses instead of brackets. They are lazy.
```python
squares = (x*x for x in range(10))
```
**Infinite generators:**
Because generators are lazy, you can create infinite sequences (e.g., Fibonacci, prime numbers) without crashing.
**Generator pipelines:**
You can chain generators: one generator reads raw data, another filters, another transforms. Each step works on one item at a time – memory usage remains constant.
**When to use generators:**
- Processing large files (reading line by line).
- Producing infinite streams (sensor readings, stock prices).
- Implementing custom iterators with low memory footprint.
- Building data processing pipelines.
---
## 3. CONTEXT MANAGERS
A **context manager** allows you to allocate and release resources automatically using the `with` statement. The classic example is file handling: `with open('file.txt') as f:` ensures the file is closed even if an exception occurs.
**Class‑based context manager:**
Implement `__enter__` and `__exit__` methods.
- `__enter__` runs when entering the `with` block; its return value is bound to the `as` variable.
- `__exit__` runs when leaving the block, even if an exception occurred. It receives exception details; returning `True` suppresses the exception.
**Using `contextlib.contextmanager`:**
For simple use cases, the `@contextlib.contextmanager` decorator turns a generator function into a context manager. Use `yield` to separate setup and teardown – everything before `yield` is setup, everything after is teardown (inside a `try`‑`finally`).
**Common applications:**
- File I/O.
- Database connections (ensure disconnection).
- Locks (threading).
- Temporary changes (e.g., modifying a setting and reverting).
- Measuring execution time.
- Rate limiting (API calls).
**Benefits:**
- Automatic cleanup (no forgotten `close()`).
- Exception safety.
- Readable, concise code.
---
## Putting It All Together
You can combine these concepts: a decorator can wrap a generator, and a context manager can provide a timing or resource‑handling wrapper around the whole pipeline. The examples in the code demonstrate:
- A `timer` decorator that also works with a generator‑based processing function.
- A `cache` decorator for memoisation.
- A `RateLimiter` context manager for API throttling.
- A `paginate_data` generator for paginating large datasets.
---
## Practice Exercises
1. Write a decorator `@log_call` that prints `"Calling function X with args Y"` before executing and `"Finished"` after.
2. Write a generator `primes()` that yields prime numbers infinitely. Use it to find the first 10 primes.
3. Write a context manager `timed_block` that prints how long the indented block took to execute.
4. Use `@contextlib.contextmanager` to create a temporary directory (create on entry, delete on exit).
5. Combine a generator that reads a large file line by line with a decorator that times the whole reading process.
These advanced features are frequently encountered in professional Python codebases. Learning them will significantly improve your ability to write efficient, clean, and reusable code.
# ADVANCED PYTHON TOPICS
import time
import contextlib
from functools import wraps
print("ADVANCED PYTHON TOPICS")
print("=" * 60)
# ========== DECORATORS ==========
print("\n1. DECORATORS")
print("-" * 30)
print("Decorators wrap functions to add functionality.")
print("Common uses: Logging, timing, authentication, caching")
# Example 1: Basic decorator
def simple_decorator(func):
"""A simple decorator that prints before and after function call"""
def wrapper():
print("Before function call")
func()
print("After function call")
return wrapper
@simple_decorator
def say_hello():
print("Hello!")
print("\nExample 1 - Basic decorator:")
say_hello()
# Example 2: Decorator with arguments
def repeat(n_times):
"""Decorator that repeats function call n times"""
def decorator(func):
@wraps(func) # Preserves function metadata
def wrapper(*args, **kwargs):
for _ in range(n_times):
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@repeat(n_times=3)
def greet(name):
print(f"Hello, {name}!")
print("\nExample 2 - Decorator with arguments:")
greet("Alice")
# Example 3: Timing decorator
def timer(func):
"""Decorator that measures function execution time"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"{func.__name__} took {end_time - start_time:.4f} seconds")
return result
return wrapper
@timer
def slow_function():
"""A function that takes some time"""
time.sleep(0.5)
return "Done"
print("\nExample 3 - Timing decorator:")
result = slow_function()
print(f"Result: {result}")
# Example 4: Caching decorator (memoization)
def cache(func):
"""Decorator that caches function results"""
cache_dict = {}
@wraps(func)
def wrapper(*args):
if args in cache_dict:
print(f"Cache hit for {args}")
return cache_dict[args]
print(f"Cache miss for {args}")
result = func(*args)
cache_dict[args] = result
return result
return wrapper
@cache
def expensive_computation(n):
"""Simulate expensive computation"""
time.sleep(0.2)
return n * n
print("\nExample 4 - Caching decorator:")
print(f"First call: {expensive_computation(5)}")
print(f"Second call (cached): {expensive_computation(5)}")
print(f"New call: {expensive_computation(10)}")
# Example 5: Authorization decorator
def requires_auth(role="user"):
"""Decorator that checks authorization"""
def decorator(func):
@wraps(func)
def wrapper(user_role, *args, **kwargs):
if user_role == role or user_role == "admin":
return func(*args, **kwargs)
else:
return f"Access denied. Required role: {role}"
return wrapper
return decorator
@requires_auth(role="admin")
def delete_user(user_id):
return f"User {user_id} deleted"
print("\nExample 5 - Authorization decorator:")
print(f"Admin try: {delete_user('admin', 'user123')}")
print(f"User try: {delete_user('user', 'user123')}")
# ========== GENERATORS ==========
print("\n\n2. GENERATORS")
print("-" * 30)
print("Generators produce values one at a time, saving memory.")
print("Use 'yield' instead of 'return'")
# Example 6: Basic generator
def count_up_to(n):
"""Generator that counts up to n"""
count = 1
while count <= n:
yield count
count += 1
print("\nExample 6 - Basic generator:")
print("Counting to 5:")
for number in count_up_to(5):
print(f" {number}")
# Example 7: Infinite generator
def fibonacci_generator():
"""Generator that produces Fibonacci sequence"""
a, b = 0, 1
while True:
yield a
a, b = b, a + b
print("\nExample 7 - Fibonacci generator (first 10):")
fib = fibonacci_generator()
for i in range(10):
print(f" Fibonacci {i}: {next(fib)}")
# Example 8: Generator expression
print("\nExample 8 - Generator expression:")
# Similar to list comprehension but lazy
squares = (x*x for x in range(10))
print("First 5 squares:")
for i in range(5):
print(f" {next(squares)}")
# Example 9: Pipeline with generators
def read_lines(filename):
"""Generator that reads file line by line"""
# Simulate file reading
lines = ["Line 1: Hello", "Line 2: World", "Line 3: Python", "Line 4: Generator"]
for line in lines:
yield line
def filter_lines(lines, keyword):
"""Generator that filters lines containing keyword"""
for line in lines:
if keyword in line:
yield line
def process_lines(lines):
"""Generator that processes lines"""
for line in lines:
yield line.upper()
print("\nExample 9 - Generator pipeline:")
# Create pipeline
lines = read_lines("sample.txt")
filtered = filter_lines(lines, "Python")
processed = process_lines(filtered)
print("Pipeline output:")
for line in processed:
print(f" {line}")
# Example 10: Generator for large datasets
def process_large_dataset(n):
"""Simulate processing large dataset"""
for i in range(n):
# Simulate processing each item
data = f"Item {i}: Data point {i*10}"
yield data
print("\nExample 10 - Large dataset processing:")
print("Processing 1 million items (in memory efficient way):")
# Only processes one item at a time
for i, item in enumerate(process_large_dataset(5)): # Using 5 for demo
print(f" {item}")
if i >= 2: # Stop early for demo
print(" ... (stopping early for demo)")
break
# ========== CONTEXT MANAGERS ==========
print("\n\n3. CONTEXT MANAGERS")
print("-" * 30)
print("Context managers handle setup and teardown automatically.")
print("Use 'with' statement for automatic cleanup.")
# Example 11: Basic context manager with class
class TimerContext:
"""Context manager for timing code blocks"""
def __enter__(self):
self.start_time = time.time()
print("Starting timer...")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.end_time = time.time()
print(f"Time elapsed: {self.end_time - self.start_time:.4f} seconds")
# Return False to propagate exceptions, True to suppress
return False
print("\nExample 11 - Context manager class:")
with TimerContext() as timer:
time.sleep(0.3)
print(" Doing some work inside context...")
# Example 12: Context manager for file handling
class SafeFileWriter:
"""Context manager for safe file writing"""
def __init__(self, filename, mode="w"):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
if exc_type is not None:
print(f"Error occurred: {exc_val}")
print(f"File '{self.filename}' closed safely")
# Don't suppress exceptions
return False
print("\nExample 12 - File context manager:")
try:
with SafeFileWriter("test.txt") as f:
f.write("Hello, Context Manager!\n")
f.write("This is automatically closed.")
# Simulate error
# raise ValueError("Test error")
print(" File written successfully")
except Exception as e:
print(f" Error: {e}")
# Example 13: Using contextlib for simpler context managers
@contextlib.contextmanager
def temporary_change(obj, attr, new_value):
"""Temporarily change an object's attribute"""
old_value = getattr(obj, attr)
setattr(obj, attr, new_value)
try:
yield
finally:
setattr(obj, attr, old_value)
class Settings:
def __init__(self):
self.mode = "light"
self.language = "en"
print("\nExample 13 - Context manager with contextlib:")
settings = Settings()
print(f"Original mode: {settings.mode}")
with temporary_change(settings, "mode", "dark"):
print(f" Inside context: {settings.mode}")
print(f"After context: {settings.mode}")
# Example 14: Database connection context manager
class DatabaseConnection:
"""Simulated database connection"""
def __init__(self, db_name):
self.db_name = db_name
self.connected = False
def connect(self):
print(f"Connecting to {self.db_name}...")
self.connected = True
return self
def disconnect(self):
if self.connected:
print(f"Disconnecting from {self.db_name}...")
self.connected = False
def execute_query(self, query):
if not self.connected:
raise ConnectionError("Not connected to database")
print(f"Executing: {query}")
return f"Results for: {query}"
class DBContextManager:
"""Context manager for database connections"""
def __init__(self, db_name):
self.db_name = db_name
self.conn = None
def __enter__(self):
self.conn = DatabaseConnection(self.db_name).connect()
return self.conn
def __exit__(self, exc_type, exc_val, exc_tb):
if self.conn:
self.conn.disconnect()
if exc_type:
print(f"Database error: {exc_val}")
return False # Don't suppress exceptions
print("\nExample 14 - Database context manager:")
with DBContextManager("mydatabase.db") as db:
result = db.execute_query("SELECT * FROM users")
print(f" {result}")
# Example 15: Combining all concepts
print("\n\n4. COMBINING ALL CONCEPTS")
print("-" * 30)
@timer
def process_with_generators(n):
"""Process data using generators with timing"""
data_gen = (i for i in range(n))
@cache
def square(x):
return x * x
results = []
for item in data_gen:
results.append(square(item))
return sum(results)
class ProcessContext:
"""Context manager for processing"""
def __enter__(self):
print("Starting processing context...")
self.start_memory = "Simulated memory check"
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("Cleaning up processing context...")
self.end_memory = "Simulated memory check"
print("Processing complete")
return False
print("\nCombined example:")
with ProcessContext():
result = process_with_generators(1000)
print(f"Result: {result}")
# Example 16: Real-world use case - API rate limiting
import time
class RateLimiter:
"""Context manager for API rate limiting"""
def __init__(self, calls_per_second=1):
self.calls_per_second = calls_per_second
self.min_interval = 1.0 / calls_per_second
self.last_call = 0
def __enter__(self):
current_time = time.time()
elapsed = current_time - self.last_call
if elapsed < self.min_interval:
sleep_time = self.min_interval - elapsed
print(f"Rate limiting: sleeping for {sleep_time:.2f} seconds")
time.sleep(sleep_time)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.last_call = time.time()
return False
def make_api_call(endpoint):
"""Simulated API call"""
print(f"Calling API: {endpoint}")
return f"Response from {endpoint}"
print("\nExample 16 - API rate limiter:")
limiter = RateLimiter(calls_per_second=2) # Max 2 calls per second
for i in range(3):
with limiter:
response = make_api_call(f"/api/data/{i}")
print(f" {response}")
# Example 17: Generator-based pagination
print("\n\n5. PRACTICAL EXAMPLE: PAGINATION WITH GENERATORS")
print("-" * 30)
def paginate_data(total_items, page_size):
"""Generator for paginating through data"""
total_pages = (total_items + page_size - 1) // page_size
for page in range(total_pages):
start = page * page_size
end = min(start + page_size, total_items)
yield {
"page": page + 1,
"start": start,
"end": end,
"data": list(range(start, end)) # Simulated data
}
print("Pagination example (20 items, 5 per page):")
for page_info in paginate_data(20, 5):
print(f"Page {page_info['page']}: Items {page_info['start']}-{page_info['end']-1}")
print(f" Data: {page_info['data']}")
→ Run this code interactively