今日练习主题：高级Python特性与元编程

今天我们将深入学习Python的高级特性，包括装饰器高级用法、上下文管理器、元类、描述符等元编程技术。

练习1：高级装饰器与函数工具

import functools
import time
from typing import Any, Callable, TypeVar, cast
from dataclasses import dataclass
from enum import Enum

# 类型变量用于泛型
F = TypeVar('F', bound=Callable[..., Any])

class CacheType(Enum):
    """缓存类型枚举"""
    LRU = "lru"
    TTl = "ttl"
    FIFO = "fifo"

def advanced_decorators():
    """高级装饰器用法"""
    print("=== 高级装饰器与函数工具 ===")
    
    # 1. 带参数的装饰器工厂
    def retry(max_attempts: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
        """
        重试装饰器
        
        Args:
            max_attempts: 最大重试次数
            delay: 重试延迟（秒）
            exceptions: 需要重试的异常类型
        """
        def decorator(func: F) -> F:
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                last_exception = None
                for attempt in range(max_attempts):
                    try:
                        return func(*args, **kwargs)
                    except exceptions as e:
                        last_exception = e
                        if attempt < max_attempts - 1:
                            print(f"尝试 {func.__name__} 失败 (第{attempt + 1}次): {e}")
                            time.sleep(delay * (attempt + 1))  # 指数退避
                        else:
                            print(f"所有 {max_attempts} 次尝试都失败了")
                raise last_exception
            return cast(F, wrapper)
        return decorator
    
    # 2. 类装饰器
    class Singleton:
        """单例装饰器"""
        def __init__(self, cls):
            self.cls = cls
            self.instance = None
            functools.update_wrapper(self, cls)
        
        def __call__(self, *args, **kwargs):
            if self.instance is None:
                self.instance = self.cls(*args, **kwargs)
            return self.instance
    
    # 3. 方法装饰器
    def method_deprecated(message: str = ""):
        """标记方法已弃用"""
        def decorator(method):
            @functools.wraps(method)
            def wrapper(self, *args, **kwargs):
                warning_msg = f"方法 {method.__name__} 已弃用"
                if message:
                    warning_msg += f": {message}"
                print(f"警告: {warning_msg}")
                return method(self, *args, **kwargs)
            return wrapper
        return decorator
    
    # 4. 属性装饰器
    def validate_range(min_val: float, max_val: float):
        """验证属性值范围"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(self, value):
                if not (min_val <= value <= max_val):
                    raise ValueError(f"值必须在 {min_val} 和 {max_val} 之间")
                return func(self, value)
            return wrapper
        return decorator
    
    # 5. 缓存装饰器
    class CacheDecorator:
        """高级缓存装饰器"""
        
        def __init__(self, maxsize: int = 128, ttl: int = 300):
            self.maxsize = maxsize
            self.ttl = ttl
            self.cache = {}
            self.access_order = []
        
        def __call__(self, func: F) -> F:
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                # 创建缓存键
                key = self._make_key(args, kwargs)
                
                # 检查缓存
                if key in self.cache:
                    result, timestamp = self.cache[key]
                    if time.time() - timestamp < self.ttl:
                        # 更新访问顺序
                        self.access_order.remove(key)
                        self.access_order.append(key)
                        return result
                    else:
                        # 缓存过期
                        del self.cache[key]
                        self.access_order.remove(key)
                
                # 执行函数
                result = func(*args, **kwargs)
                
                # 更新缓存
                self.cache[key] = (result, time.time())
                self.access_order.append(key)
                
                # 清理过期缓存
                self._cleanup()
                
                return result
            
            return cast(F, wrapper)
        
        def _make_key(self, args, kwargs):
            """创建缓存键"""
            key = (args, tuple(sorted(kwargs.items())))
            return hash(key)
        
        def _cleanup(self):
            """清理缓存"""
            current_time = time.time()
            
            # 移除过期缓存
            expired_keys = [
                key for key, (_, timestamp) in self.cache.items()
                if current_time - timestamp >= self.ttl
            ]
            for key in expired_keys:
                del self.cache[key]
                self.access_order.remove(key)
            
            # 如果依旧超过最大大小，移除最久未使用的
            while len(self.cache) > self.maxsize and self.access_order:
                oldest_key = self.access_order.pop(0)
                if oldest_key in self.cache:
                    del self.cache[oldest_key]
    
    # 使用示例
    @Singleton
    class DatabaseConnection:
        """数据库连接单例"""
        def __init__(self):
            print("创建数据库连接")
            self.connected = True
        
        def query(self, sql: str):
            return f"执行: {sql}"
    
    class Calculator:
        def __init__(self):
            self._value = 0
        
        @property
        def value(self):
            return self._value
        
        @value.setter
        @validate_range(0, 100)
        def value(self, val):
            self._value = val
        
        @retry(max_attempts=3, delay=0.5, exceptions=(ValueError,))
        def risky_operation(self, x):
            """有风险的操作"""
            if x < 0:
                raise ValueError("不能处理负数")
            return x * 2
        
        @method_deprecated("请使用 new_method 取代")
        def old_method(self):
            return "旧方法"
        
        @CacheDecorator(maxsize=10, ttl=60)
        def expensive_calculation(self, n):
            """昂贵的计算"""
            print(f"执行昂贵计算: {n}")
            time.sleep(0.1)
            return sum(i * i for i in range(n))
    
    # 演示
    print("1. 单例模式演示:")
    db1 = DatabaseConnection()
    db2 = DatabaseConnection()
    print(f"db1 is db2: {db1 is db2}")
    
    print("
2. 重试机制演示:")
    calc = Calculator()
    try:
        result = calc.risky_operation(-5)
    except ValueError as e:
        print(f"捕获异常: {e}")
    
    result = calc.risky_operation(5)
    print(f"成功结果: {result}")
    
    print("
3. 属性验证演示:")
    calc.value = 50
    print(f"设置值: {calc.value}")
    try:
        calc.value = 150
    except ValueError as e:
        print(f"验证错误: {e}")
    
    print("
4. 缓存演示:")
    # 第一次调用
    start_time = time.time()
    result1 = calc.expensive_calculation(1000)
    time1 = time.time() - start_time
    
    # 第二次调用（应该从缓存获取）
    start_time = time.time()
    result2 = calc.expensive_calculation(1000)
    time2 = time.time() - start_time
    
    print(f"第一次执行时间: {time1:.4f}s")
    print(f"第二次执行时间: {time2:.4f}s")
    print(f"结果一样: {result1 == result2}")
    print(f"加速比: {time1/time2:.1f}x")
    
    print("
5. 弃用警告演示:")
    calc.old_method()

# 运行高级装饰器示例
advanced_decorators()

练习2：元类与描述符

def metaclasses_and_descriptors():
    """元类与描述符高级用法"""
    print("
=== 元类与描述符 ===")
    
    # 1. 描述符基础
    class ValidatedAttribute:
        """验证描述符"""
        
        def __init__(self, min_value=None, max_value=None, type_=None):
            self.min_value = min_value
            self.max_value = max_value
            self.type_ = type_
            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):
            self._validate(value)
            instance.__dict__[self.name] = value
        
        def _validate(self, value):
            """验证属性值"""
            if self.type_ and not isinstance(value, self.type_):
                raise TypeError(f"{self.name} 必须是 {self.type_.__name__} 类型")
            
            if self.min_value is not None and value < self.min_value:
                raise ValueError(f"{self.name} 不能小于 {self.min_value}")
            
            if self.max_value is not None and value > self.max_value:
                raise ValueError(f"{self.name} 不能大于 {self.max_value}")
    
    # 2. 延迟加载描述符
    class LazyProperty:
        """延迟加载描述符"""
        
        def __init__(self, func):
            self.func = func
            self.name = None
            functools.update_wrapper(self, func)
        
        def __set_name__(self, owner, name):
            self.name = name
        
        def __get__(self, instance, owner):
            if instance is None:
                return self
            
            # 计算并缓存结果
            value = self.func(instance)
            instance.__dict__[self.name] = value
            return value
    
    # 3. 观察者描述符
    class ObservableAttribute:
        """可观察属性描述符"""
        
        def __init__(self):
            self.observers = []
            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):
            old_value = instance.__dict__.get(self.name)
            instance.__dict__[self.name] = value
            
            # 通知观察者
            for callback in self.observers:
                callback(instance, self.name, old_value, value)
        
        def add_observer(self, callback):
            """添加观察者"""
            self.observers.append(callback)
    
    # 4. 自定义元类
    class ValidationMeta(type):
        """验证元类"""
        
        def __new__(cls, name, bases, attrs):
            # 收集验证规则
            validators = {}
            for attr_name, attr_value in attrs.items():
                if isinstance(attr_value, ValidatedAttribute):
                    validators[attr_name] = attr_value
            
            # 创建类
            new_class = super().__new__(cls, name, bases, attrs)
            new_class._validators = validators
            return new_class
        
        def __init__(cls, name, bases, attrs):
            super().__init__(name, bases, attrs)
            
            # 重写 __setattr__ 以进行验证
            original_setattr = cls.__setattr__
            
            def validating_setattr(instance, name, value):
                # 如果是验证属性，先验证
                if name in cls._validators:
                    cls._validators[name]._validate(value)
                original_setattr(instance, name, value)
            
            cls.__setattr__ = validating_setattr
    
    # 5. 注册表元类
    class RegistryMeta(type):
        """注册表元类"""
        
        _registry = {}
        
        def __new__(cls, name, bases, attrs):
            new_class = super().__new__(cls, name, bases, attrs)
            
            # 注册类
            if 'name' in attrs:
                cls._registry[attrs['name']] = new_class
            
            return new_class
        
        @classmethod
        def get_class(cls, name):
            """根据名称获取类"""
            return cls._registry.get(name)
    
    # 6. 使用元类和描述符的类
    class Person(metaclass=ValidationMeta):
        """使用验证元类的Person类"""
        
        # 使用验证描述符
        age = ValidatedAttribute(min_value=0, max_value=150, type_=int)
        name = ValidatedAttribute(type_=str)
        email = ValidatedAttribute()
        
        # 使用可观察属性
        salary = ObservableAttribute()
        
        def __init__(self, name, age, email, salary=0):
            self.name = name
            self.age = age
            self.email = email
            self.salary = salary
        
        @LazyProperty
        def formatted_info(self):
            """延迟加载的属性"""
            print("计算格式化信息...")
            time.sleep(0.5)  # 模拟昂贵计算
            return f"{self.name} ({self.age}岁) - {self.email}"
    
    # 7. 使用注册表元类的类
    class Animal(metaclass=RegistryMeta):
        """动物基类"""
        
        def speak(self):
            raise NotImplementedError
    
    class Dog(Animal):
        name = "dog"
        
        def speak(self):
            return "汪汪!"
    
    class Cat(Animal):
        name = "cat"
        
        def speak(self):
            return "喵喵!"
    
    # 演示代码
    print("1. 验证描述符演示:")
    try:
        person = Person("张三", 25, "zhang@example.com")
        print(f"创建成功: {person.name}, {person.age}岁")
        
        # 测试验证
        try:
            person.age = -5
        except ValueError as e:
            print(f"验证错误: {e}")
        
        try:
            person.age = "二十五"
        except TypeError as e:
            print(f"类型错误: {e}")
            
    except Exception as e:
        print(f"创建失败: {e}")
    
    print("
2. 延迟加载属性演示:")
    person = Person("李四", 30, "li@example.com")
    print("第一次访问 formatted_info:")
    info1 = person.formatted_info
    print(f"信息: {info1}")
    
    print("第二次访问 formatted_info (应该从缓存获取):")
    info2 = person.formatted_info
    print(f"信息: {info2}")
    print(f"一样对象: {info1 is info2}")
    
    print("
3. 可观察属性演示:")
    def salary_change_callback(instance, attr_name, old_value, new_value):
        print(f"工资变化: {old_value} -> {new_value}")
    
    # 添加观察者
    Person.salary.add_observer(salary_change_callback)
    
    person.salary = 5000
    person.salary = 6000
    
    print("
4. 注册表元类演示:")
    dog_class = RegistryMeta.get_class("dog")
    cat_class = RegistryMeta.get_class("cat")
    
    if dog_class:
        dog = dog_class()
        print(f"狗叫: {dog.speak()}")
    
    if cat_class:
        cat = cat_class()
        print(f"猫叫: {cat.speak()}")
    
    print(f"注册表中的类: {list(RegistryMeta._registry.keys())}")
    
    # 8. 动态类创建
    print("
5. 动态类创建:")
    
    def create_dynamic_class(class_name, attributes):
        """动态创建类"""
        return type(class_name, (), attributes)
    
    # 动态创建类
    DynamicClass = create_dynamic_class(
        "DynamicClass",
        {
            'x': 10,
            'y': 20,
            'get_sum': lambda self: self.x + self.y,
            '__str__': lambda self: f"DynamicClass(x={self.x}, y={self.y})"
        }
    )
    
    obj = DynamicClass()
    print(f"动态对象: {obj}")
    print(f"求和: {obj.get_sum()}")
    
    # 9. 方法注入
    print("
6. 方法注入:")
    
    def inject_method(cls, method_name, method):
        """向类注入方法"""
        setattr(cls, method_name, method)
    
    # 向Person类注入新方法
    def get_annual_salary(self):
        return self.salary * 12
    
    inject_method(Person, 'get_annual_salary', get_annual_salary)
    
    person = Person("王五", 35, "wang@example.com", salary=8000)
    print(f"年薪: {person.get_annual_salary()}")

# 运行元类与描述符示例
metaclasses_and_descriptors()

综合练习：创建高级框架

def create_advanced_framework():
    """创建高级框架综合练习"""
    print("
=== 创建高级框架 ===")
    
    # 1. 声明式API框架
    class DeclarativeMeta(type):
        """声明式元类"""
        
        def __new__(cls, name, bases, attrs):
            # 收集字段信息
            fields = {}
            for attr_name, attr_value in attrs.items():
                if isinstance(attr_value, Field):
                    fields[attr_name] = attr_value
                    attr_value.name = attr_name
            
            # 创建类
            new_class = super().__new__(cls, name, bases, attrs)
            new_class._declared_fields = fields
            
            # 添加验证方法
            new_class.validate = cls._create_validate_method(fields)
            
            return new_class
        
        @staticmethod
        def _create_validate_method(fields):
            """创建验证方法"""
            def validate(self):
                errors = {}
                for field_name, field in fields.items():
                    try:
                        value = getattr(self, field_name)
                        field.validate(value)
                    except ValueError as e:
                        errors[field_name] = str(e)
                return errors
            return validate
    
    class Field:
        """字段基类"""
        
        def __init__(self, required=True, default=None):
            self.required = required
            self.default = default
            self.name = None
        
        def validate(self, value):
            """验证字段值"""
            if self.required and value is None:
                raise ValueError(f"字段 {self.name} 是必需的")
            return True
    
    class StringField(Field):
        """字符串字段"""
        
        def __init__(self, min_length=0, max_length=100, **kwargs):
            super().__init__(**kwargs)
            self.min_length = min_length
            self.max_length = max_length
        
        def validate(self, value):
            super().validate(value)
            
            if value is None:
                return True
            
            if not isinstance(value, str):
                raise ValueError(f"字段 {self.name} 必须是字符串")
            
            if len(value) < self.min_length:
                raise ValueError(f"字段 {self.name} 长度不能小于 {self.min_length}")
            
            if len(value) > self.max_length:
                raise ValueError(f"字段 {self.name} 长度不能大于 {self.max_length}")
            
            return True
    
    class IntegerField(Field):
        """整数字段"""
        
        def __init__(self, min_value=None, max_value=None, **kwargs):
            super().__init__(**kwargs)
            self.min_value = min_value
            self.max_value = max_value
        
        def validate(self, value):
            super().validate(value)
            
            if value is None:
                return True
            
            if not isinstance(value, int):
                raise ValueError(f"字段 {self.name} 必须是整数")
            
            if self.min_value is not None and value < self.min_value:
                raise ValueError(f"字段 {self.name} 不能小于 {self.min_value}")
            
            if self.max_value is not None and value > self.max_value:
                raise ValueError(f"字段 {self.name} 不能大于 {self.max_value}")
            
            return True
    
    # 2. 使用声明式API
    class UserForm(metaclass=DeclarativeMeta):
        """用户表单"""
        
        username = StringField(min_length=3, max_length=20, required=True)
        email = StringField(required=True)
        age = IntegerField(min_value=0, max_value=150, required=True)
        bio = StringField(required=False, max_length=500)
        
        def __init__(self, **kwargs):
            for field_name in self._declared_fields:
                value = kwargs.get(field_name, self._declared_fields[field_name].default)
                setattr(self, field_name, value)
        
        def is_valid(self):
            """检查表单是否有效"""
            return len(self.validate()) == 0
        
        def __str__(self):
            fields_str = ", ".join(
                f"{name}={getattr(self, name)}" 
                for name in self._declared_fields
            )
            return f"UserForm({fields_str})"
    
    # 3. 插件系统
    class Plugin:
        """插件基类"""
        
        def __init__(self, name):
            self.name = name
        
        def execute(self, context):
            raise NotImplementedError
    
    class PluginManager:
        """插件管理器"""
        
        def __init__(self):
            self.plugins = {}
            self.hooks = {}
        
        def register_plugin(self, plugin):
            """注册插件"""
            self.plugins[plugin.name] = plugin
            print(f"注册插件: {plugin.name}")
        
        def add_hook(self, hook_name, plugin_name):
            """添加钩子"""
            if hook_name not in self.hooks:
                self.hooks[hook_name] = []
            self.hooks[hook_name].append(plugin_name)
        
        def execute_hook(self, hook_name, context):
            """执行钩子"""
            if hook_name not in self.hooks:
                return context
            
            for plugin_name in self.hooks[hook_name]:
                if plugin_name in self.plugins:
                    plugin = self.plugins[plugin_name]
                    context = plugin.execute(context)
                    print(f"执行插件 {plugin_name}，结果: {context}")
            
            return context
    
    # 4. 具体插件实现
    class LoggingPlugin(Plugin):
        """日志插件"""
        
        def execute(self, context):
            print(f"[日志] 处理数据: {context}")
            return context
    
    class ValidationPlugin(Plugin):
        """验证插件"""
        
        def execute(self, context):
            if isinstance(context, dict) and 'data' in context:
                context['validated'] = True
                print(f"[验证] 数据已验证")
            return context
    
    class TransformationPlugin(Plugin):
        """转换插件"""
        
        def execute(self, context):
            if isinstance(context, dict) and 'data' in context:
                context['data'] = context['data'].upper()
                print(f"[转换] 数据已转换")
            return context
    
    # 演示代码
    print("1. 声明式API演示:")
    
    # 有效表单
    valid_form = UserForm(username="john_doe", email="john@example.com", age=25)
    print(f"表单: {valid_form}")
    print(f"是否有效: {valid_form.is_valid()}")
    print(f"验证错误: {valid_form.validate()}")
    
    # 无效表单
    print("
无效表单示例:")
    invalid_form = UserForm(username="ab", email="invalid", age=-5)
    print(f"表单: {invalid_form}")
    print(f"是否有效: {invalid_form.is_valid()}")
    print(f"验证错误: {invalid_form.validate()}")
    
    print("
2. 插件系统演示:")
    
    # 创建插件管理器
    plugin_manager = PluginManager()
    
    # 注册插件
    plugins = [
        LoggingPlugin("logger"),
        ValidationPlugin("validator"),
        TransformationPlugin("transformer")
    ]
    
    for plugin in plugins:
        plugin_manager.register_plugin(plugin)
    
    # 设置钩子链
    plugin_manager.add_hook("process_data", "logger")
    plugin_manager.add_hook("process_data", "validator")
    plugin_manager.add_hook("process_data", "transformer")
    
    # 执行数据处理
    data_context = {"data": "hello world", "source": "test"}
    print(f"
原始数据: {data_context}")
    
    result = plugin_manager.execute_hook("process_data", data_context)
    print(f"最终结果: {result}")
    
    # 5. 依赖注入容器
    print("
3. 依赖注入容器:")
    
    class DIContainer:
        """依赖注入容器"""
        
        def __init__(self):
            self.services = {}
            self.singletons = {}
        
        def register(self, service_name, factory, singleton=False):
            """注册服务"""
            self.services[service_name] = (factory, singleton)
            print(f"注册服务: {service_name} (singleton: {singleton})")
        
        def get(self, service_name):
            """获取服务实例"""
            if service_name not in self.services:
                raise ValueError(f"服务未注册: {service_name}")
            
            factory, singleton = self.services[service_name]
            
            if singleton:
                if service_name not in self.singletons:
                    self.singletons[service_name] = factory(self)
                return self.singletons[service_name]
            else:
                return factory(self)
    
    # 使用依赖注入
    container = DIContainer()
    
    # 注册服务
    container.register("database", lambda c: "DatabaseConnection", singleton=True)
    container.register("logger", lambda c: "LoggerService", singleton=True)
    container.register("user_service", 
                      lambda c: f"UserService(db={c.get('database')}, log={c.get('logger')})")
    
    # 获取服务
    db = container.get("database")
    logger = container.get("logger")
    user_service = container.get("user_service")
    
    print(f"数据库: {db}")
    print(f"日志器: {logger}")
    print(f"用户服务: {user_service}")
    
    # 验证单例
    db2 = container.get("database")
    print(f"数据库单例验证: {db is db2}")

# 运行高级框架示例
create_advanced_framework()

print("
" + "="*60)
print("第二十一天学习完成！")
print("="*60)
print("今天学习了：")
print("✓ 高级装饰器模式和工厂")
print("✓ 描述符协议和属性管理")
print("✓ 元类编程和类定制")
print("✓ 动态类创建和方法注入")
print("✓ 声明式API设计")
print("✓ 插件系统和依赖注入")
print("✓ 框架设计模式")

学习总结：

高级装饰器：

• 带参数的装饰器工厂
• 类装饰器和单例模式
• 方法装饰器和属性验证
• 高级缓存和重试机制

元编程技术：

• 描述符协议和属性管理
• 元类定制和类创建
• 动态类和方法操作
• 注册表模式和插件系统

框架设计：

• 声明式API设计
• 依赖注入容器
• 插件架构
• 验证框架

这些高级特性让Python具备了强劲的元编程能力，可以创建灵活的框架和DSL（领域特定语言）。明天我们将学习更多实际应用！