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="alice@example.com", age=30)print(user)  # User(name='Alice', email='alice@example.com', 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': 'alice@example.com'} # Testauth_request = Request(authenticated=True)print(get_user_data(auth_request, 123))# {"data": {"id": 123, "name": "Alice", "email": "alice@example.com"}, "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

Bài 3: Generators và Iterators - Lazy evaluation, memory-efficient iteration, và generator expressions! 🚀

Remember:

Class decorators modify classes
Decorator classes use __call__
@property for computed attributes
@classmethod for factories
Stack decorators thoughtfully! 🎯

#class-decorators

Decorators - Class Decorators & Advanced Patterns