Python学习--MetaClass(元类)详解

"""元类是创建类的类。关系链：元类（metaclass） → 类（class） → 实例（instance）Python中，一切都是对象：- 整数、字符串、列表等是对象- 类也是对象（type的实例）- type是type自身的实例（最特殊的元类）"""# type是所有类的默认元类print(f"type的类型: {type(type)}")          # <class 'type'>print(f"type是自身的实例: {type(type) istype}")  # True# 类的创建过程class MyClass:    passobj = MyClass()print(f"实例的类型: {type(obj)}")           # <class '__main__.MyClass'>print(f"类的类型: {type(MyClass)}")        # <class 'type'>print(f"类的元类: {MyClass.__class__}")    # <class 'type'># 证明类也是对象class_list = [MyClass, str, list, dict]for cls in class_list:    print(f"{cls.__name__} 的类型是 {type(cls).__name__}")

"""类的创建流程：1. Python解释器遇到class关键字2. 收集类名、基类、类体代码3. 调用元类的__new__()方法创建类4. 调用元类的__init__()方法初始化类5. 将类对象绑定到类名实例的创建流程：1. 调用类的__call__()方法（即元类的__call__()）2. 调用类的__new__()方法3. 调用类的__init__()方法"""# 使用type手动创建类（模拟class关键字的工作）def __init__(self, name):    self.name = namedef say_hello(self):    return f"Hello, I'm {self.name}"# 等效于 class Person: ...Person = type('Person', (), {    '__init__': __init__,    'say_hello': say_hello,    'species': '人类'})# 使用person = Person("张三")print(person.say_hello())  # Hello, I'm 张三print(person.species)      # 人类

class SimpleMeta(type):    """最简单的自定义元类"""    def __new__(mcs, name, bases, attrs):        """        创建类对象        参数:            mcs: 元类自身（cls）            name: 类名            bases: 基类元组            attrs: 类属性字典        """        print(f"【元类__new__】创建类: {name}")        print(f"  基类: {bases}")        print(f"  属性: {list(attrs.keys())}")        # 调用type的__new__创建类        cls = super().__new__(mcs, name, bases, attrs)        # 可以修改或添加类属性        cls.created_by = 'SimpleMeta'        cls.creation_time = 'now'        return cls    def __init__(cls, name, bases, attrs):        """        初始化类对象        参数中的cls是刚刚创建的类对象        """        print(f"【元类__init__】初始化类: {name}")        super().__init__(name, bases, attrs)    def __call__(cls, *args, **kwargs):        """        创建类的实例时调用        这控制了实例的创建过程        """        print(f"【元类__call__】创建 {cls.__name__} 的实例")        print(f"  参数: args={args}, kwargs={kwargs}")        # 1. 调用类的__new__        instance = cls.__new__(cls)        # 2. 调用类的__init__        if instance is not None and hasattr(instance, '__init__'):            instance.__init__(*args, **kwargs)        # 可以修改实例        instance.created_by_meta = True        return instance# 使用自定义元类class MyClass(metaclass=SimpleMeta):    """使用SimpleMeta作为元类"""    def __init__(self, value):        self.value = value        print(f"【类__init__】初始化实例，value={value}")    def __new__(cls, *args, **kwargs):        print(f"【类__new__】创建实例")        return super().__new__(cls)print("=" * 50)print("开始创建类...")print("-" * 50)# 类创建时调用元类的__new__和__init__# 已经在类定义时执行了print("-" * 50)print("开始创建实例...")print("-" * 50)# 实例创建时调用元类的__call__obj = MyClass(42)print("-" * 50)print(f"实例属性: value={obj.value}, created_by_meta={obj.created_by_meta}")print(f"类属性: created_by={MyClass.created_by}")

class PluginMeta(type):    """插件注册元类：自动注册所有子类"""    # 注册表    _registry = {}    def __new__(mcs, name, bases, attrs):        # 创建类        cls = super().__new__(mcs, name, bases, attrs)        # 跳过基类（不注册抽象的Plugin类）        if bases != (object,) and 'Plugin' in [base.__name__ for base in bases]:            # 注册类            plugin_name = attrs.get('plugin_name', name.lower())            mcs._registry[plugin_name] = cls            print(f"✅ 注册插件: {plugin_name} -> {name}")        return cls    @classmethod    def get_plugin(cls, name):        """获取插件"""        return cls._registry.get(name)    @classmethod    def list_plugins(cls):        """列出所有插件"""        return list(cls._registry.keys())# 基类：使用PluginMeta作为元类class Plugin(metaclass=PluginMeta):    """插件基类"""    plugin_name = None    def execute(self, *args, **kwargs):        raise NotImplementedError("子类必须实现execute方法")# 插件类会自动注册class ImageProcessor(Plugin):    """图片处理器插件"""    plugin_name = 'image_processor'    def execute(self, image_path):        return f"处理图片: {image_path}"class TextAnalyzer(Plugin):    """文本分析插件"""    plugin_name = 'text_analyzer'    def execute(self, text):        return f"分析文本: {text[:50]}..."class DataExporter(Plugin):    """数据导出插件"""    # 使用类名作为插件名（小写）    def execute(self, data):        return f"导出数据: {len(data)}条记录"# 使用print("\n已注册的插件:")for name in PluginMeta.list_plugins():    print(f"  - {name}")print("\n使用插件:")image_plugin = PluginMeta.get_plugin('image_processor')()result = image_plugin.execute("/path/to/image.jpg")print(f"  图片处理结果: {result}")text_plugin = PluginMeta.get_plugin('text_analyzer')()result = text_plugin.execute("这是一段很长的文本内容，需要进行分析和处理...")print(f"  文本分析结果: {result}")data_plugin = PluginMeta.get_plugin('dataexporter')()  # 使用类名小写result = data_plugin.execute([1, 2, 3, 4, 5])print(f"  数据导出结果: {result}")

class Field:    """数据库字段描述符"""    def __init__(self, field_type, primary_key=False, nullable=True, default=None):        self.field_type = field_type        self.primary_key = primary_key        self.nullable = nullable        self.default = default        self.name = None  # 会在元类中设置    def __get__(self, instance, owner):        if instance is None:            return self        return instance.__dict__.get(self.name, self.default)    def __set__(self, instance, value):        if value is None and not self.nullable:            raise ValueError(f"{self.name} 不能为空")        if value is not None and not isinstance(value, self.field_type):            raise TypeError(f"{self.name} 必须是 {self.field_type.__name__} 类型")        instance.__dict__[self.name] = valueclass ModelMeta(type):    """模型元类：自动处理字段和表映射"""    def __new__(mcs, name, bases, attrs):        # 跳过基类Model        if name == 'Model':            return super().__new__(mcs, name, bases, attrs)        print(f"【ModelMeta】创建模型类: {name}")        # 收集字段        fields = {}        primary_key = None        for attr_name, attr_value in attrs.items():            if isinstance(attr_value, Field):                # 设置字段名                attr_value.name = attr_name                fields[attr_name] = attr_value                # 检查主键                if attr_value.primary_key:                    if primary_key is not None:                        raise ValueError("只能有一个主键字段")                    primary_key = attr_name        # 移除字段描述符，防止成为类属性        for field_name in fields:            attrs.pop(field_name)        # 添加元数据        attrs['_fields'] = fields        attrs['_table_name'] = name.lower() + 's'  # 简单复数化        attrs['_primary_key'] = primary_key        # 添加通用方法        attrs['__init__'] = mcs.create_init(fields)        attrs['save'] = mcs.create_save()        attrs['delete'] = mcs.create_delete()        attrs['to_dict'] = lambda self: {name: getattr(self, name) for name in fields}        # 添加类方法        attrs['create_table_sql'] = classmethod(mcs.create_table_sql)        attrs['get'] = classmethod(mcs.create_get())        # 创建类        cls = super().__new__(mcs, name, bases, attrs)        print(f"  表名: {cls._table_name}")        print(f"  字段: {list(fields.keys())}")        print(f"  主键: {primary_key}")        return cls    @staticmethod    def create_init(fields):        """创建__init__方法"""        def init(self, **kwargs):            for field_name, field in fields.items():                value = kwargs.get(field_name, field.default)                setattr(self, field_name, value)        return init    @staticmethod    def create_save():        """创建save方法"""        def save(self):            # 模拟保存到数据库            data = self.to_dict()            if self._primary_key and getattr(self, self._primary_key) is None:                # 插入新记录                print(f"INSERT INTO {self._table_name} VALUES {tuple(data.values())}")            else:                # 更新记录                print(f"UPDATE {self._table_name} SET {data} WHERE {self._primary_key}={getattr(self, self._primary_key)}")            return True        return save    @staticmethod    def create_delete():        """创建delete方法"""        def delete(self):            if self._primary_key and getattr(self, self._primary_key) is not None:                print(f"DELETE FROM {self._table_name} WHERE {self._primary_key}={getattr(self, self._primary_key)}")                return True            return False        return delete    @staticmethod    def create_table_sql(cls):        """创建建表SQL"""        columns = []        for field_name, field in cls._fields.items():            column_def = f"{field_name}{field.field_type.__name__.upper()}"            if field.primary_key:                column_def += " PRIMARY KEY"            if not field.nullable:                column_def += " NOT NULL"            if field.default is not None:                column_def += f" DEFAULT {repr(field.default)}"            columns.append(column_def)        sql = f"CREATE TABLE IF NOT EXISTS {cls._table_name} (\n  "        sql += ",\n  ".join(columns)        sql += "\n);"        return sql    @staticmethod    def create_get():        """创建get类方法"""        @classmethod        def get(cls, pk_value):            # 模拟从数据库查询            print(f"SELECT * FROM {cls._table_name} WHERE {cls._primary_key}={pk_value}")            # 这里应该返回实例，为了示例返回None            return None        return get# 基类class Model(metaclass=ModelMeta):    """模型基类"""    pass# 使用元类自动创建模型class User(Model):    """用户模型"""    id = Field(int, primary_key=True)    username = Field(str, nullable=False)    email = Field(str)    age = Field(int, default=18)    is_active = Field(bool, default=True)class Product(Model):    """产品模型"""    id = Field(int, primary_key=True)    name = Field(str, nullable=False)    price = Field(float, nullable=False)    stock = Field(int, default=0)# 使用模型print("\n" + "="*50)print("使用ORM模型:")print("="*50)# 查看建表SQLprint("\nUser表SQL:")print(User.create_table_sql())print("\nProduct表SQL:")print(Product.create_table_sql())# 创建实例print("\n创建用户:")user = User(username="张三", email="zhang@example.com", age=25)print(f"用户数据: {user.to_dict()}")# 保存print("\n保存用户:")user.save()# 设置ID并更新user.id = 1print("\n更新用户:")user.save()# 删除print("\n删除用户:")user.delete()# 查询print("\n查询用户:")User.get(1)

class RouteMeta(type):    """路由注册元类：自动注册路由"""    _routes = {}    def __new__(mcs, name, bases, attrs):        # 创建类        cls = super().__new__(mcs, name, bases, attrs)        # 跳过基类        if bases != (object,):            # 注册路由            prefix = attrs.get('route_prefix', '/' + name.lower())            print(f"🚀 注册控制器: {name} (前缀: {prefix})")            # 收集路由方法            for attr_name, attr_value in attrs.items():                if callable(attr_value) and hasattr(attr_value, '_route'):                    # 获取路由信息                    route_info = attr_value._route                    path = prefix + route_info['path']                    methods = route_info['methods']                    # 注册路由                    full_path = f"{methods}{path}"                    mcs._routes[full_path] = {                        'controller': cls,                        'method': attr_name,                        'handler': attr_value                    }                    print(f"  → 注册路由: {full_path} -> {attr_name}")        return cls    @classmethod    def dispatch(cls, method, path):        """分发请求"""        key = f"{method}{path}"        route = cls._routes.get(key)        if route:            # 创建控制器实例并调用方法            controller = route['controller']()            return route['handler'](controller)        return {"error": "Route not found", "path": path}    @classmethod    def list_routes(cls):        """列出所有路由"""        return cls._routes# 路由装饰器def route(path, methods=['GET']):    """路由装饰器"""    def decorator(func):        func._route = {'path': path, 'methods': ' '.join(methods)}        return func    return decorator# 控制器基类class Controller(metaclass=RouteMeta):    """控制器基类"""    route_prefix = ''# 用户控制器class UserController(Controller):    route_prefix = '/users'    @route('/')    def list_users(self):        return {"action": "list_users", "data": ["user1", "user2", "user3"]}    @route('/<int:user_id>', methods=['GET'])    def get_user(self, user_id):        return {"action": "get_user", "user_id": user_id}    @route('/<int:user_id>', methods=['PUT'])    def update_user(self, user_id):        return {"action": "update_user", "user_id": user_id}    @route('/', methods=['POST'])    def create_user(self):        return {"action": "create_user"}# 产品控制器class ProductController(Controller):    route_prefix = '/products'    @route('/')    def list_products(self):        return {"action": "list_products", "data": ["product1", "product2"]}    @route('/<int:product_id>')    def get_product(self, product_id):        return {"action": "get_product", "product_id": product_id}# 使用print("\n" + "="*50)print("路由系统:")print("="*50)print("\n已注册的路由:")routes = RouteMeta.list_routes()for route_path, info in routes.items():    print(f"  {route_path} -> {info['controller'].__name__}.{info['method']}")print("\n模拟HTTP请求:")print("-" * 30)# 模拟请求分发test_requests = [    ('GET', '/users'),    ('GET', '/users/123'),    ('PUT', '/users/123'),    ('POST', '/users'),    ('GET', '/products'),    ('GET', '/products/456'),    ('GET', '/not-found'),]for method, path in test_requests:    print(f"\n{method}{path}:")    result = RouteMeta.dispatch(method, path)    print(f"  结果: {result}")

class SingletonMeta(type):    """单例模式元类"""    _instances = {}    def __call__(cls, *args, **kwargs):        # 如果类还没有实例，创建并存储        if cls not in cls._instances:            print(f"🆕 创建 {cls.__name__} 的单例实例")            instance = super().__call__(*args, **kwargs)            cls._instances[cls] = instance        else:            print(f"♻️  使用 {cls.__name__} 的现有实例")        return cls._instances[cls]    @classmethod    def clear_instances(mcs):        """清除所有单例实例（用于测试）"""        mcs._instances.clear()        print("🧹 已清除所有单例实例")class DatabaseConnection(metaclass=SingletonMeta):    """数据库连接（单例）"""    def __init__(self, host='localhost', port=3306):        self.host = host        self.port = port        self.connection = None        print(f"  初始化数据库连接: {host}:{port}")    def connect(self):        if not self.connection:            self.connection = f"Connection to {self.host}:{self.port}"            print(f"  建立连接: {self.connection}")        return self.connection    def query(self, sql):        print(f"  执行查询: {sql}")        return f"Result from {self.host}: {sql}"class ConfigManager(metaclass=SingletonMeta):    """配置管理器（单例）"""    def __init__(self, config_file='config.json'):        self.config_file = config_file        self.config = self.load_config()        print(f"  加载配置文件: {config_file}")    def load_config(self):        # 模拟加载配置        return {"app_name": "MyApp", "version": "1.0.0"}    def get(self, key, default=None):        return self.config.get(key, default)# 使用单例print("单例模式演示:")print("=" * 50)print("\n1. 数据库连接:")db1 = DatabaseConnection("db1.example.com", 3306)db1.connect()print("\n尝试创建第二个实例（应该是同一个）:")db2 = DatabaseConnection("db1.example.com", 3306)db2.connect()print(f"\ndb1 is db2: {db1 is db2}")print(f"db1.host: {db1.host}, db2.host: {db2.host}")print("\n2. 配置管理器:")config1 = ConfigManager()print(f"应用名: {config1.get('app_name')}")print("\n尝试创建第二个配置管理器:")config2 = ConfigManager("other_config.json")print(f"配置文件: {config2.config_file} (应该是第一次的文件名)")print(f"\nconfig1 is config2: {config1 is config2}")print("\n3. 创建不同类的单例:")print("即使都是单例，不同类也是不同的实例")db = DatabaseConnection()config = ConfigManager()print(f"db is config: {db is config}")# 清理（仅用于演示）print("\n清理单例实例:")SingletonMeta.clear_instances()print("\n再次创建数据库连接（会创建新实例）:")db3 = DatabaseConnection("new.example.com", 5432)print(f"db1 is db3: {db1 is db3}")