Bài 2: Decorators - Class Decorators & Advanced Patterns
Mục Tiêu Bài Học
Sau khi hoàn thành bài này, bạn sẽ:
- ✅ Tạo class decorators
- ✅ Sử dụng decorator classes (callable objects)
- ✅ Hiểu built-in decorators (@property, @staticmethod, @classmethod)
- ✅ Tạo decorators cho methods
- ✅ Áp dụng advanced decorator patterns
- ✅ Combine multiple decorator techniques
Class Decorators
Class decorator nhận class làm input và return class mới hoặc modified class.
Basic Class Decorator
def add_repr(cls): """Add __repr__ method to class.""" def __repr__(self): attrs = ', '.join(f"{k}={v!r}" for k, v in self.__dict__.items()) return f"{cls.__name__}({attrs})" cls.__repr__ = __repr__ return cls @add_reprclass Person: def __init__(self, name, age): self.name = name self.age = age @add_reprclass Product: def __init__(self, name, price): self.name = name self.price = price # Usageperson = Person("Alice", 30)print(person) # Person(name='Alice', age=30) product = Product("Laptop", 999.99)print(product) # Product(name='Laptop', price=999.99)
Class Decorator với Parameters
def singleton(cls): """Make class a singleton.""" instances = {} def get_instance(*args, **kwargs): if cls not in instances: instances[cls] = cls(*args, **kwargs) return instances[cls] return get_instance @singletonclass Database: def __init__(self): self.connection = "Connected to DB" print("Database initialized") # Usagedb1 = Database() # Database initializeddb2 = Database() # No output (same instance) print(db1 is db2) # Trueprint(db1.connection) # Connected to DB
Dataclass-like Decorator
def dataclass(cls): """ Simple dataclass decorator. Auto-generates __init__, __repr__, __eq__. """ # Get class annotations annotations = getattr(cls, '__annotations__', {}) # Generate __init__ def __init__(self, **kwargs): for field_name, field_type in annotations.items(): if field_name in kwargs: setattr(self, field_name, kwargs[field_name]) else: raise TypeError(f"Missing required field: {field_name}") # Generate __repr__ def __repr__(self): fields = ', '.join( f"{name}={getattr(self, name)!r}" for name in annotations ) return f"{cls.__name__}({fields})" # Generate __eq__ def __eq__(self, other): if not isinstance(other, cls): return False return all( getattr(self, name) == getattr(other, name) for name in annotations ) # Attach methods cls.__init__ = __init__ cls.__repr__ = __repr__ cls.__eq__ = __eq__ return cls @dataclassclass Point: x: int y: int @dataclassclass User: name: str email: str age: int # Usagep1 = Point(x=10, y=20)print(p1) # Point(x=10, y=20) p2 = Point(x=10, y=20)print(p1 == p2) # True user = User(name="Alice", email="[email protected]", age=30)print(user) # User(name='Alice', email='[email protected]', age=30)
Registry Decorator
# Plugin registryPLUGINS = {} def register_plugin(name): """Register plugin by name.""" def decorator(cls): PLUGINS[name] = cls cls.plugin_name = name return cls return decorator @register_plugin('image')class ImageProcessor: def process(self, data): return f"Processing image: {data}" @register_plugin('video')class VideoProcessor: def process(self, data): return f"Processing video: {data}" @register_plugin('audio')class AudioProcessor: def process(self, data): return f"Processing audio: {data}" # Usagedef process_file(file_type, data): """Process file using registered plugin.""" if file_type not in PLUGINS: raise ValueError(f"Unknown file type: {file_type}") processor_class = PLUGINS[file_type] processor = processor_class() return processor.process(data) print(process_file('image', 'photo.jpg'))# Processing image: photo.jpg print(process_file('video', 'movie.mp4'))# Processing video: movie.mp4 print(PLUGINS)# {'image': <class 'ImageProcessor'>, 'video': <class 'VideoProcessor'>, ...}
Decorator Classes (Callable Objects)
Class có method __call__ có thể làm decorator.
Basic Decorator Class
class CountCalls: """Count function calls using class.""" def __init__(self, func): self.func = func self.count = 0 def __call__(self, *args, **kwargs): self.count += 1 print(f"Call {self.count} to {self.func.__name__}") return self.func(*args, **kwargs) @CountCallsdef greet(name): return f"Hello, {name}!" # Usageprint(greet("Alice")) # Call 1 to greetprint(greet("Bob")) # Call 2 to greetprint(greet("Charlie")) # Call 3 to greet print(f"Total calls: {greet.count}") # Total calls: 3
Decorator Class với Parameters
class Retry: """Retry decorator as a class.""" def __init__(self, max_attempts=3, delay=1): self.max_attempts = max_attempts self.delay = delay def __call__(self, func): """Return wrapper function.""" from functools import wraps import time @wraps(func) def wrapper(*args, **kwargs): attempts = 0 while attempts < self.max_attempts: try: return func(*args, **kwargs) except Exception as e: attempts += 1 if attempts >= self.max_attempts: print(f"Failed after {self.max_attempts} attempts") raise print(f"Attempt {attempts} failed: {e}") print(f"Retrying in {self.delay}s...") time.sleep(self.delay) return wrapper @Retry(max_attempts=3, delay=2)def unstable_function(): """Simulate unstable function.""" import random if random.random() < 0.7: raise ValueError("Random failure") return "Success!" # Usageresult = unstable_function()print(result)
Stateful Decorator Class
class RateLimiter: """Rate limiter with state tracking.""" def __init__(self, max_calls, period): self.max_calls = max_calls self.period = period self.calls = {} # Track calls per function def __call__(self, func): from functools import wraps import time @wraps(func) def wrapper(*args, **kwargs): now = time.time() # Initialize if first call if func not in self.calls: self.calls[func] = [] # Remove old calls self.calls[func] = [ call_time for call_time in self.calls[func] if call_time > now - self.period ] # Check limit if len(self.calls[func]) >= self.max_calls: wait_time = self.period - (now - self.calls[func][0]) raise RuntimeError( f"Rate limit: {self.max_calls} calls per {self.period}s. " f"Wait {wait_time:.1f}s" ) # Record and execute self.calls[func].append(now) return func(*args, **kwargs) return wrapper # Create shared limiterlimiter = RateLimiter(max_calls=3, period=10) @limiterdef api_call_1(): return "API 1 response" @limiterdef api_call_2(): return "API 2 response" # Usageimport time for i in range(4): try: print(api_call_1()) time.sleep(2) except RuntimeError as e: print(f"Error: {e}")
Built-in Decorators
1. @property
class Circle: """Circle with property decorators.""" def __init__(self, radius): self._radius = radius @property def radius(self): """Get radius.""" return self._radius @radius.setter def radius(self, value): """Set radius with validation.""" if value <= 0: raise ValueError("Radius must be positive") self._radius = value @property def diameter(self): """Computed property.""" return self._radius * 2 @property def area(self): """Computed area.""" import math return math.pi * self._radius ** 2 # Usagecircle = Circle(5) print(circle.radius) # 5print(circle.diameter) # 10print(circle.area) # 78.53981633974483 circle.radius = 10print(circle.area) # 314.1592653589793 # circle.radius = -5 # ValueError: Radius must be positive# circle.diameter = 20 # AttributeError: can't set attribute
2. @staticmethod và @classmethod
class MathUtils: """Math utilities with different method types.""" class_name = "MathUtils" def __init__(self, value): self.value = value # Instance method def add(self, other): """Add to instance value.""" return self.value + other # Static method - no access to instance or class @staticmethod def is_even(number): """Check if number is even.""" return number % 2 == 0 @staticmethod def factorial(n): """Calculate factorial.""" if n <= 1: return 1 return n * MathUtils.factorial(n - 1) # Class method - access to class @classmethod def from_string(cls, value_str): """Create instance from string.""" return cls(int(value_str)) @classmethod def get_class_name(cls): """Get class name.""" return cls.class_name # Usagemath = MathUtils(10)print(math.add(5)) # 15 # Static methods - can call without instanceprint(MathUtils.is_even(10)) # Trueprint(MathUtils.is_even(11)) # Falseprint(MathUtils.factorial(5)) # 120 # Class methodsmath2 = MathUtils.from_string("42")print(math2.value) # 42print(MathUtils.get_class_name()) # MathUtils
3. Real-world @classmethod
from datetime import datetime class Person: """Person with alternative constructors.""" def __init__(self, name, age): self.name = name self.age = age @classmethod def from_birth_year(cls, name, birth_year): """Create from birth year.""" age = datetime.now().year - birth_year return cls(name, age) @classmethod def from_dict(cls, data): """Create from dictionary.""" return cls(data['name'], data['age']) def __repr__(self): return f"Person(name={self.name!r}, age={self.age})" # Usageperson1 = Person("Alice", 30)print(person1) # Person(name='Alice', age=30) person2 = Person.from_birth_year("Bob", 1990)print(person2) # Person(name='Bob', age=35) person3 = Person.from_dict({'name': 'Charlie', 'age': 25})print(person3) # Person(name='Charlie', age=25)
Method Decorators
Decorators cho class methods require special handling.
Method Decorator with Self
from functools import wraps def log_method(func): """Log method calls.""" @wraps(func) def wrapper(self, *args, **kwargs): class_name = self.__class__.__name__ print(f"[{class_name}.{func.__name__}] Called") result = func(self, *args, **kwargs) print(f"[{class_name}.{func.__name__}] Returned {result}") return result return wrapper class Calculator: """Calculator with logged methods.""" @log_method def add(self, a, b): return a + b @log_method def multiply(self, a, b): return a * b # Usagecalc = Calculator()result = calc.add(5, 3)# [Calculator.add] Called# [Calculator.add] Returned 8 result = calc.multiply(4, 5)# [Calculator.multiply] Called# [Calculator.multiply] Returned 20
Descriptor-based Method Decorator
class LogMethod: """Method decorator using descriptor protocol.""" def __init__(self, func): self.func = func self.call_count = {} def __get__(self, obj, objtype=None): """Support instance method binding.""" if obj is None: return self from functools import wraps @wraps(self.func) def wrapper(*args, **kwargs): # Track calls per instance obj_id = id(obj) self.call_count[obj_id] = self.call_count.get(obj_id, 0) + 1 print(f"Call {self.call_count[obj_id]} to {self.func.__name__}") return self.func(obj, *args, **kwargs) return wrapper class Service: """Service with tracked methods.""" @LogMethod def process(self, data): return f"Processed: {data}" @LogMethod def validate(self, data): return len(data) > 0 # Usageservice1 = Service()service1.process("data1") # Call 1 to processservice1.process("data2") # Call 2 to processservice1.validate("test") # Call 1 to validate service2 = Service()service2.process("data3") # Call 1 to process (separate instance)
Advanced Decorator Patterns
1. Context-aware Decorator
from functools import wrapsimport threading # Thread-local storage for context_context = threading.local() def with_context(**context_kwargs): """Execute function with context.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): # Save old context old_context = getattr(_context, 'data', {}).copy() # Set new context if not hasattr(_context, 'data'): _context.data = {} _context.data.update(context_kwargs) try: return func(*args, **kwargs) finally: # Restore old context _context.data = old_context return wrapper return decorator def get_context(key, default=None): """Get value from context.""" if not hasattr(_context, 'data'): return default return _context.data.get(key, default) @with_context(user='admin', role='superuser')def admin_operation(): """Operation with admin context.""" user = get_context('user') role = get_context('role') return f"Executing as {user} ({role})" @with_context(user='guest', role='viewer')def guest_operation(): """Operation with guest context.""" user = get_context('user') role = get_context('role') return f"Executing as {user} ({role})" # Usageprint(admin_operation()) # Executing as admin (superuser)print(guest_operation()) # Executing as guest (viewer)print(get_context('user')) # None (context cleared)
2. Conditional Execution Decorator
from functools import wraps def run_if(condition_func): """Only run function if condition is True.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): if condition_func(): return func(*args, **kwargs) else: print(f"Skipping {func.__name__}: condition not met") return None return wrapper return decorator # Configurationconfig = { 'debug_mode': True, 'maintenance_mode': False,} @run_if(lambda: config['debug_mode'])def debug_log(message): """Log only in debug mode.""" print(f"[DEBUG] {message}") @run_if(lambda: not config['maintenance_mode'])def process_request(request): """Process only when not in maintenance.""" return f"Processing: {request}" # Usagedebug_log("This will print")# [DEBUG] This will print result = process_request("user_request")print(result) # Processing: user_request config['maintenance_mode'] = Trueresult = process_request("another_request")# Skipping process_request: condition not metprint(result) # None
3. Chained Validation Decorators
from functools import wraps def validate_not_none(func): """Validate result is not None.""" @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if result is None: raise ValueError(f"{func.__name__} returned None") return result return wrapper def validate_positive(func): """Validate result is positive number.""" @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if not isinstance(result, (int, float)) or result <= 0: raise ValueError(f"{func.__name__} must return positive number") return result return wrapper def validate_type(expected_type): """Validate return type.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if not isinstance(result, expected_type): raise TypeError( f"{func.__name__} must return {expected_type.__name__}" ) return result return wrapper return decorator # Chain validators@validate_not_none@validate_positive@validate_type(int)def calculate_age(birth_year): """Calculate age with validation.""" from datetime import datetime return datetime.now().year - birth_year # Usagetry: age = calculate_age(1990) print(f"Age: {age}") # Age: 35 # This will fail validation # calculate_age(2030) # ValueError: positive numberexcept ValueError as e: print(f"Error: {e}")
Practical Examples
1. API Endpoint Decorator Suite
from functools import wraps class APIEndpoint: """Decorator suite for API endpoints.""" @staticmethod def json_response(func): """Convert result to JSON response.""" @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) import json return json.dumps({'data': result, 'success': True}) return wrapper @staticmethod def require_auth(func): """Require authentication.""" @wraps(func) def wrapper(request, *args, **kwargs): if not getattr(request, 'authenticated', False): import json return json.dumps({'error': 'Unauthorized', 'success': False}) return func(request, *args, **kwargs) return wrapper @staticmethod def rate_limit(max_calls=10): """Rate limit endpoint.""" calls = [] def decorator(func): @wraps(func) def wrapper(*args, **kwargs): import time now = time.time() # Clean old calls calls[:] = [t for t in calls if t > now - 60] if len(calls) >= max_calls: import json return json.dumps({'error': 'Rate limit exceeded', 'success': False}) calls.append(now) return func(*args, **kwargs) return wrapper return decorator # Mock request objectclass Request: def __init__(self, authenticated=False): self.authenticated = authenticated # Usage@APIEndpoint.rate_limit(max_calls=5)@APIEndpoint.require_auth@APIEndpoint.json_responsedef get_user_data(request, user_id): """Get user data endpoint.""" return {'id': user_id, 'name': 'Alice', 'email': '[email protected]'} # Testauth_request = Request(authenticated=True)print(get_user_data(auth_request, 123))# {"data": {"id": 123, "name": "Alice", "email": "[email protected]"}, "success": true} unauth_request = Request(authenticated=False)print(get_user_data(unauth_request, 123))# {"error": "Unauthorized", "success": false}
2. Caching System with Multiple Strategies
from functools import wrapsimport time class Cache: """Cache decorator with different strategies.""" @staticmethod def lru(maxsize=128): """LRU (Least Recently Used) cache.""" cache = {} access_times = {} def decorator(func): @wraps(func) def wrapper(*args): key = args # Hit if key in cache: access_times[key] = time.time() return cache[key] # Miss result = func(*args) # Evict if full if len(cache) >= maxsize: lru_key = min(access_times, key=access_times.get) del cache[lru_key] del access_times[lru_key] # Store cache[key] = result access_times[key] = time.time() return result return wrapper return decorator @staticmethod def ttl(seconds=60): """TTL (Time To Live) cache.""" cache = {} timestamps = {} def decorator(func): @wraps(func) def wrapper(*args): key = args now = time.time() # Check if cached and not expired if key in cache and now - timestamps[key] < seconds: return cache[key] # Compute and cache result = func(*args) cache[key] = result timestamps[key] = now return result return wrapper return decorator # Usage@Cache.lru(maxsize=3)def expensive_lru(n): """Expensive computation with LRU cache.""" print(f"Computing {n}...") time.sleep(0.1) return n ** 2 @Cache.ttl(seconds=5)def expensive_ttl(n): """Expensive computation with TTL cache.""" print(f"Computing {n}...") time.sleep(0.1) return n ** 2 # Test LRUprint(expensive_lru(1)) # Computing 1...print(expensive_lru(2)) # Computing 2...print(expensive_lru(1)) # Cachedprint(expensive_lru(3)) # Computing 3...print(expensive_lru(4)) # Computing 4... (evicts oldest) # Test TTLprint(expensive_ttl(5)) # Computing 5...print(expensive_ttl(5)) # Cachedtime.sleep(6)print(expensive_ttl(5)) # Computing 5... (expired)
3. Performance Profiler
from functools import wrapsimport timeimport tracemalloc class Profile: """Profiling decorators.""" @staticmethod def time_and_memory(func): """Profile time and memory.""" @wraps(func) def wrapper(*args, **kwargs): # Start tracking tracemalloc.start() start_time = time.time() # Execute result = func(*args, **kwargs) # Get metrics end_time = time.time() current, peak = tracemalloc.get_traced_memory() tracemalloc.stop() # Report print(f"\n{'='*50}") print(f"Function: {func.__name__}") print(f"Time: {end_time - start_time:.4f} seconds") print(f"Memory (current): {current / 1024 / 1024:.2f} MB") print(f"Memory (peak): {peak / 1024 / 1024:.2f} MB") print(f"{'='*50}\n") return result return wrapper @Profile.time_and_memorydef process_large_data(): """Process large dataset.""" # Create large list data = [i ** 2 for i in range(1000000)] # Some processing result = sum(data) return result # Usageresult = process_large_data()# ==================================================# Function: process_large_data# Time: 0.1234 seconds# Memory (current): 38.15 MB# Memory (peak): 76.29 MB# ==================================================
Best Practices
# 1. Use class decorators for class modifications@singletonclass Configuration: pass # 2. Use decorator classes for stateful decoratorsclass CallCounter: def __init__(self, func): self.func = func self.count = 0 def __call__(self, *args, **kwargs): self.count += 1 return self.func(*args, **kwargs) # 3. Combine @property with validationclass Product: @property def price(self): return self._price @price.setter def price(self, value): if value < 0: raise ValueError("Price cannot be negative") self._price = value # 4. Use @classmethod for alternative constructorsclass Person: @classmethod def from_birth_year(cls, name, year): return cls(name, 2025 - year) # 5. Stack decorators in logical order@timer # Outermost - measure everything@logger # Log the call@validate # Validate inputs@cache # Cache resultsdef complex_operation(x): return x ** 2
Bài Tập Thực Hành
Bài 1: Singleton with Thread-Safety
Tạo singleton decorator thread-safe using locks.
Bài 2: Auto-save Decorator
Tạo class decorator auto-save object state after method calls.
Bài 3: Lazy Property
Tạo lazy property decorator chỉ compute once khi first access.
Bài 4: Method Chaining
Tạo decorator enable method chaining (return self).
Bài 5: Async Decorator
Tạo decorator convert sync function to async.
Tóm Tắt
✅ Class Decorators: Modify classes - add methods, attributes
✅ Decorator Classes: Use __call__ để làm decorator
✅ @property: Managed attributes với getters/setters
✅ @staticmethod: Methods không cần instance/class
✅ @classmethod: Methods có access to class
✅ Method Decorators: Special handling cho methods
✅ Advanced Patterns: Context, conditional, validation chains
Kết Luận Cả 2 Parts
Part 1: Function decorators, basic patterns, practical examples
Part 2: Class decorators, decorator classes, built-in decorators, advanced patterns
Key Takeaways:
- Decorators modify behavior without changing code
- Use @wraps to preserve metadata
- Class decorators powerful for modifying classes
- Decorator classes good for stateful decorators
- Built-in decorators (@property, @classmethod) are essential
- Combine decorators for complex behavior
Bài Tiếp Theo
Remember:
- Class decorators modify classes
- Decorator classes use
__call__ - @property for computed attributes
- @classmethod for factories
- Stack decorators thoughtfully! 🎯