Skip to content

Latest commit

 

History

History
706 lines (573 loc) · 24 KB

File metadata and controls

706 lines (573 loc) · 24 KB

Customizing Subclass Behavior with init_subclass in Python

Slide 1: Understanding init_subclass in Python

Class parameterization during inheritance is a powerful feature introduced in Python 3.6 through init_subclass. This method allows parent classes to customize subclass creation behavior by intercepting and modifying the subclass definition process, providing a cleaner alternative to metaclasses.

# Base class that customizes subclass creation
class ConfigurableBase:
    @classmethod
    def __init_subclass__(cls, prefix="default_", **kwargs):
        super().__init_subclass__(**kwargs)
        # Customize all methods in subclass with prefix
        for name, method in cls.__dict__.items():
            if callable(method) and not name.startswith('__'):
                setattr(cls, f"{prefix}{name}", method)

# Example subclass with custom prefix
class Worker(ConfigurableBase, prefix="worker_"):
    def process(self):
        return "Processing data"

# Usage demonstration
worker = Worker()
print(worker.worker_process())  # Output: Processing data

Slide 2: Metaclass Implementation Pre-Python 3.6

Before init_subclass, developers relied on metaclasses to achieve similar subclass customization. This approach requires understanding Python's type system and class creation process, making it more complex but offering greater control over class creation.

class ParameterizedMeta(type):
    def __new__(cls, name, bases, namespace, **kwargs):
        # Customize class creation based on parameters
        prefix = kwargs.get('prefix', 'default_')
        
        # Create new namespace with modified methods
        new_namespace = {}
        for key, value in namespace.items():
            if callable(value) and not key.startswith('__'):
                new_namespace[f"{prefix}{key}"] = value
            else:
                new_namespace[key] = value
                
        return super().__new__(cls, name, bases, new_namespace)
    
    def __init__(cls, name, bases, namespace, **kwargs):
        super().__init__(name, bases, namespace)

# Example usage with metaclass
class Worker(metaclass=ParameterizedMeta, prefix='worker_'):
    def process(self):
        return "Processing data"

worker = Worker()
print(worker.worker_process())  # Output: Processing data

Slide 3: Advanced Method Decoration with init_subclass

The init_subclass method enables sophisticated method decoration patterns, allowing parent classes to automatically enhance or modify subclass methods. This approach maintains clean inheritance hierarchies while adding powerful functionality to all derived classes.

class LoggedBase:
    @classmethod
    def __init_subclass__(cls, log_methods=True, **kwargs):
        super().__init_subclass__(**kwargs)
        if log_methods:
            # Wrap all methods with logging functionality
            for name, method in cls.__dict__.items():
                if callable(method) and not name.startswith('__'):
                    setattr(cls, name, LoggedBase.log_decorator(method))
    
    @staticmethod
    def log_decorator(method):
        def wrapper(*args, **kwargs):
            print(f"Calling method: {method.__name__}")
            result = method(*args, **kwargs)
            print(f"Method {method.__name__} returned: {result}")
            return result
        return wrapper

class DataProcessor(LoggedBase):
    def process_data(self, data):
        return f"Processed: {data}"

# Usage example
processor = DataProcessor()
processor.process_data("sample")

Slide 4: Dynamic Interface Enforcement

init_subclass can be used to enforce interface requirements dynamically during class definition. This pattern ensures that subclasses implement required methods while providing helpful error messages during development.

class InterfaceEnforcer:
    _required_methods = set()
    
    @classmethod
    def __init_subclass__(cls, required_methods=None, **kwargs):
        super().__init_subclass__(**kwargs)
        if required_methods:
            cls._required_methods = set(required_methods)
            
        # Verify all required methods are implemented
        missing_methods = cls._required_methods - set(cls.__dict__.keys())
        if missing_methods:
            raise TypeError(
                f"Can't instantiate abstract class {cls.__name__} "
                f"with missing methods: {', '.join(missing_methods)}"
            )

class DataHandler(InterfaceEnforcer, required_methods=['load', 'save']):
    def load(self): 
        return "Loading data"
    
    def save(self):
        return "Saving data"

# This will raise TypeError due to missing methods
try:
    class BadHandler(InterfaceEnforcer, required_methods=['load', 'save']):
        def load(self):
            pass
except TypeError as e:
    print(e)

Slide 5: Parameterized Validation Framework

This implementation demonstrates how init_subclass can be used to create a robust validation framework where validation rules are defined through class parameters, enabling flexible and reusable data validation patterns.

class Validator:
    @classmethod
    def __init_subclass__(cls, validators=None, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._validators = validators or {}
        
        # Create validation methods dynamically
        for field, rules in cls._validators.items():
            setattr(cls, f"validate_{field}", 
                   cls._create_validator(field, rules))
    
    @staticmethod
    def _create_validator(field, rules):
        def validator(self, value):
            for rule, params in rules.items():
                if rule == 'min_length' and len(value) < params:
                    raise ValueError(
                        f"{field} must be at least {params} characters"
                    )
                if rule == 'max_length' and len(value) > params:
                    raise ValueError(
                        f"{field} must be at most {params} characters"
                    )
            return True
        return validator

class UserValidator(Validator, validators={
    'username': {'min_length': 3, 'max_length': 20},
    'password': {'min_length': 8, 'max_length': 30}
}):
    pass

# Usage example
validator = UserValidator()
try:
    validator.validate_username("ab")
except ValueError as e:
    print(e)  # Output: username must be at least 3 characters

Slide 6: Factory Pattern Using init_subclass

The init_subclass method enables elegant implementation of the factory pattern, allowing automatic registration of subclasses. This approach eliminates the need for manual registration and provides a centralized creation mechanism for related classes.

class ServiceFactory:
    _services = {}
    
    @classmethod
    def __init_subclass__(cls, service_type=None, **kwargs):
        super().__init_subclass__(**kwargs)
        if service_type:
            ServiceFactory._services[service_type] = cls
    
    @staticmethod
    def create(service_type, *args, **kwargs):
        if service_type not in ServiceFactory._services:
            raise ValueError(f"Unknown service type: {service_type}")
        return ServiceFactory._services[service_type](*args, **kwargs)

class EmailService(ServiceFactory, service_type="email"):
    def send(self, message):
        return f"Sending email: {message}"

class SMSService(ServiceFactory, service_type="sms"):
    def send(self, message):
        return f"Sending SMS: {message}"

# Usage example
email_service = ServiceFactory.create("email")
print(email_service.send("Hello!"))  # Output: Sending email: Hello!

Slide 7: Attribute Validation Framework

This implementation creates a framework for automatic attribute validation in classes. It demonstrates how init_subclass can be used to implement descriptor-like behavior with class-level configuration.

class ValidatedAttribute:
    def __init__(self, validation_func):
        self.validation_func = validation_func
        self.name = None
    
    def __set_name__(self, owner, name):
        self.name = name
    
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get(self.name)
    
    def __set__(self, instance, value):
        if self.validation_func(value):
            instance.__dict__[self.name] = value

class AttributeValidator:
    @classmethod
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        validations = kwargs.get('validations', {})
        for attr, validator in validations.items():
            setattr(cls, attr, ValidatedAttribute(validator))

# Example usage
class Person(AttributeValidator, validations={
    'age': lambda x: isinstance(x, int) and 0 <= x <= 150,
    'name': lambda x: isinstance(x, str) and len(x) > 0
}):
    pass

# Test the validation
person = Person()
person.age = 25  # Valid
try:
    person.age = -1  # Invalid
except ValueError:
    print("Invalid age")

Slide 8: Dynamic Method Generation

This advanced implementation shows how init_subclass can be used to dynamically generate methods based on class attributes, creating a powerful and flexible API generation system.

class APIEndpoint:
    @classmethod
    def __init_subclass__(cls, endpoints=None, **kwargs):
        super().__init_subclass__(**kwargs)
        if endpoints:
            for name, config in endpoints.items():
                method = cls._create_endpoint_method(name, config)
                setattr(cls, name, method)
    
    @staticmethod
    def _create_endpoint_method(name, config):
        def endpoint_method(self, **kwargs):
            # Simulate API call
            method = config.get('method', 'GET')
            path = config.get('path', f'/{name}')
            return {
                'method': method,
                'path': path,
                'params': kwargs,
                'response': f'Response from {name}'
            }
        endpoint_method.__name__ = name
        return endpoint_method

class UserAPI(APIEndpoint, endpoints={
    'get_user': {'method': 'GET', 'path': '/users/{id}'},
    'create_user': {'method': 'POST', 'path': '/users'},
    'update_user': {'method': 'PUT', 'path': '/users/{id}'}
}):
    pass

# Usage example
api = UserAPI()
print(api.get_user(id=1))
print(api.create_user(name="John", email="john@example.com"))

Slide 9: Configurable Serialization

This implementation demonstrates how init_subclass can be used to create a flexible serialization framework that automatically handles different data types and formats based on class configuration.

from datetime import datetime
import json

class Serializable:
    @classmethod
    def __init_subclass__(cls, fields=None, date_format="%Y-%m-%d", **kwargs):
        super().__init_subclass__(**kwargs)
        cls._fields = fields or []
        cls._date_format = date_format
        
        # Create serialization methods
        cls.to_dict = cls._create_to_dict()
        cls.from_dict = classmethod(cls._create_from_dict())
    
    @staticmethod
    def _create_to_dict():
        def to_dict(self):
            result = {}
            for field in self._fields:
                value = getattr(self, field)
                if isinstance(value, datetime):
                    value = value.strftime(self._date_format)
                result[field] = value
            return result
        return to_dict
    
    @staticmethod
    def _create_from_dict():
        def from_dict(cls, data):
            processed_data = {}
            for field in cls._fields:
                if field in data:
                    value = data[field]
                    if field.endswith('_date'):
                        value = datetime.strptime(value, cls._date_format)
                    processed_data[field] = value
            return cls(**processed_data)
        return from_dict

class User(Serializable, fields=['name', 'email', 'created_date']):
    def __init__(self, name, email, created_date):
        self.name = name
        self.email = email
        self.created_date = created_date

# Usage example
user = User("John", "john@example.com", datetime.now())
user_dict = user.to_dict()
print(json.dumps(user_dict, indent=2))

Slide 10: Real-World Example: Database ORM Implementation

A practical implementation of an Object-Relational Mapping (ORM) system using init_subclass for automatic table creation and field validation. This example demonstrates how to build a lightweight database abstraction layer.

import sqlite3
from datetime import datetime

class Field:
    def __init__(self, field_type, required=True):
        self.field_type = field_type
        self.required = required
        self.name = None

    def __set_name__(self, owner, name):
        self.name = name

    def validate(self, value):
        if value is None and self.required:
            raise ValueError(f"{self.name} is required")
        if value is not None and not isinstance(value, self.field_type):
            raise TypeError(f"{self.name} must be of type {self.field_type}")
        return value

class Model:
    _connection = sqlite3.connect(':memory:')
    
    @classmethod
    def __init_subclass__(cls, table_name=None, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._table_name = table_name or cls.__name__.lower()
        cls._fields = {
            name: field for name, field in cls.__dict__.items()
            if isinstance(field, Field)
        }
        cls._create_table()
    
    @classmethod
    def _create_table(cls):
        fields = []
        for name, field in cls._fields.items():
            field_type = 'TEXT' if field.field_type in (str, datetime) else 'INTEGER'
            nullable = '' if field.required else 'NULL'
            fields.append(f"{name} {field_type} {nullable}")
        
        query = f"""
        CREATE TABLE IF NOT EXISTS {cls._table_name} (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            {', '.join(fields)}
        )
        """
        cls._connection.execute(query)
        cls._connection.commit()

    def save(self):
        fields = []
        values = []
        for name, field in self.__class__._fields.items():
            value = getattr(self, name, None)
            field.validate(value)
            fields.append(name)
            values.append(
                value.isoformat() if isinstance(value, datetime) else value
            )
        
        placeholders = ','.join(['?' for _ in fields])
        query = f"""
        INSERT INTO {self._table_name} 
        ({','.join(fields)}) VALUES ({placeholders})
        """
        cursor = self._connection.execute(query, values)
        self._connection.commit()
        return cursor.lastrowid

# Example usage
class User(Model, table_name='users'):
    name = Field(str)
    age = Field(int)
    created_at = Field(datetime)

# Create and save a user
user = User()
user.name = "Alice"
user.age = 30
user.created_at = datetime.now()
user_id = user.save()

Slide 11: Real-World Example: Event-Driven Architecture

Implementation of an event-driven system using init_subclass for automatic event handler registration and management, demonstrating practical application in large-scale applications.

from typing import Callable, Dict, List
import inspect
from datetime import datetime

class EventHandler:
    def __init__(self, event_type: str):
        self.event_type = event_type

    def __call__(self, func: Callable):
        func._event_type = self.event_type
        return func

class EventSystem:
    _handlers: Dict[str, List[Callable]] = {}
    
    @classmethod
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        
        # Register all methods decorated with EventHandler
        for name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
            if hasattr(method, '_event_type'):
                event_type = method._event_type
                if event_type not in cls._handlers:
                    cls._handlers[event_type] = []
                cls._handlers[event_type].append(method)
    
    @classmethod
    def emit(cls, event_type: str, **data):
        if event_type not in cls._handlers:
            return
        
        event_data = {
            'type': event_type,
            'timestamp': datetime.now().isoformat(),
            'data': data
        }
        
        for handler in cls._handlers[event_type]:
            handler(cls, event_data)

class UserSystem(EventSystem):
    def __init__(self):
        self.users = {}

    @EventHandler("user_created")
    def log_user_creation(self, event_data):
        print(f"User created at {event_data['timestamp']}")
        print(f"Data: {event_data['data']}")

    @EventHandler("user_created")
    def send_welcome_email(self, event_data):
        user_data = event_data['data']
        print(f"Sending welcome email to {user_data['email']}")

    @EventHandler("user_deleted")
    def cleanup_user_data(self, event_data):
        user_id = event_data['data']['user_id']
        print(f"Cleaning up data for user {user_id}")

    def create_user(self, email: str, name: str):
        user_id = len(self.users) + 1
        self.users[user_id] = {'email': email, 'name': name}
        self.emit('user_created', user_id=user_id, email=email, name=name)
        return user_id

# Usage example
user_system = UserSystem()
user_system.create_user("alice@example.com", "Alice")

Slide 12: Performance Monitoring Decorator System

This implementation showcases how init_subclass can be used to create a sophisticated performance monitoring system that automatically tracks method execution times and resource usage across inherited classes.

import time
import functools
import statistics
from typing import Dict, List
import psutil

class PerformanceMonitor:
    _metrics: Dict[str, List[float]] = {}
    
    @classmethod
    def __init_subclass__(cls, monitor_methods=None, **kwargs):
        super().__init_subclass__(**kwargs)
        methods_to_monitor = monitor_methods or []
        
        for method_name in methods_to_monitor:
            if hasattr(cls, method_name):
                original_method = getattr(cls, method_name)
                wrapped_method = cls._create_monitored_method(
                    original_method, method_name
                )
                setattr(cls, method_name, wrapped_method)
    
    @classmethod
    def _create_monitored_method(cls, method, method_name):
        @functools.wraps(method)
        def wrapper(*args, **kwargs):
            start_time = time.perf_counter()
            start_memory = psutil.Process().memory_info().rss / 1024 / 1024
            
            try:
                result = method(*args, **kwargs)
                success = True
            except Exception as e:
                success = False
                raise e
            finally:
                end_time = time.perf_counter()
                end_memory = psutil.Process().memory_info().rss / 1024 / 1024
                
                metrics = {
                    'execution_time': end_time - start_time,
                    'memory_usage': end_memory - start_memory,
                    'success': success
                }
                
                if method_name not in cls._metrics:
                    cls._metrics[method_name] = []
                cls._metrics[method_name].append(metrics)
            
            return result
        return wrapper
    
    @classmethod
    def get_performance_stats(cls, method_name):
        if method_name not in cls._metrics:
            return None
        
        metrics = cls._metrics[method_name]
        times = [m['execution_time'] for m in metrics]
        memory = [m['memory_usage'] for m in metrics]
        successes = [m['success'] for m in metrics]
        
        return {
            'avg_time': statistics.mean(times),
            'max_time': max(times),
            'min_time': min(times),
            'avg_memory': statistics.mean(memory),
            'success_rate': sum(successes) / len(successes) * 100,
            'total_calls': len(metrics)
        }

# Example usage
class DataProcessor(PerformanceMonitor, monitor_methods=['process_data']):
    def process_data(self, data_size):
        # Simulate data processing
        time.sleep(0.1)  # Simulate work
        return [i * 2 for i in range(data_size)]

# Test the performance monitoring
processor = DataProcessor()
for _ in range(5):
    processor.process_data(1000)

stats = DataProcessor.get_performance_stats('process_data')
print("Performance Statistics:")
print(f"Average execution time: {stats['avg_time']:.3f} seconds")
print(f"Memory usage: {stats['avg_memory']:.2f} MB")
print(f"Success rate: {stats['success_rate']}%")
print(f"Total calls: {stats['total_calls']}")

Slide 13: Adaptive Configuration System

An implementation of an adaptive configuration system that uses init_subclass to manage hierarchical settings with inheritance and environment-specific overrides.

import os
import json
from typing import Any, Dict, Optional

class ConfigurationBase:
    _configs: Dict[str, Dict[str, Any]] = {}
    
    @classmethod
    def __init_subclass__(cls, 
                         config_path: Optional[str] = None,
                         env_prefix: Optional[str] = None,
                         **kwargs):
        super().__init_subclass__(**kwargs)
        
        cls._config_path = config_path
        cls._env_prefix = env_prefix or cls.__name__.upper() + '_'
        
        # Load configuration hierarchy
        cls._load_config()
        cls._create_properties()
    
    @classmethod
    def _load_config(cls):
        # Load from file if provided
        if cls._config_path and os.path.exists(cls._config_path):
            with open(cls._config_path, 'r') as f:
                cls._configs[cls.__name__] = json.load(f)
        else:
            cls._configs[cls.__name__] = {}
        
        # Override with environment variables
        for key in cls._configs[cls.__name__].keys():
            env_var = f"{cls._env_prefix}{key.upper()}"
            if env_var in os.environ:
                cls._configs[cls.__name__][key] = os.environ[env_var]
    
    @classmethod
    def _create_properties(cls):
        for key in cls._configs[cls.__name__].keys():
            def make_getter(k):
                def getter(self):
                    return self._configs[self.__class__.__name__][k]
                return getter
            
            def make_setter(k):
                def setter(self, value):
                    self._configs[self.__class__.__name__][k] = value
                return setter
            
            prop = property(make_getter(key), make_setter(key))
            setattr(cls, key, prop)

# Example usage
class DatabaseConfig(ConfigurationBase,
                    config_path="db_config.json",
                    env_prefix="DB_"):
    pass

class APIConfig(ConfigurationBase,
               config_path="api_config.json",
               env_prefix="API_"):
    pass

# Create example configuration file
with open("db_config.json", "w") as f:
    json.dump({
        "host": "localhost",
        "port": 5432,
        "username": "admin"
    }, f)

# Use the configuration
db_config = DatabaseConfig()
print(f"Database host: {db_config.host}")
print(f"Database port: {db_config.port}")

# Override with environment variable
os.environ["DB_PORT"] = "5433"
db_config2 = DatabaseConfig()
print(f"New database port: {db_config2.port}")

Slide 14: Additional Resources