Python元类在框架设计中的高级应用

元类（Metaclass）是Python中最强大的特性之一，它允许开发者在类创建时进行干预和定制。本文将深入探讨元类在框架设计中的高级应用。

元类基础概念

元类是类的类，它控制类的创建过程。在Python中，type是所有类的默认元类。

• type(obj) 返回对象的类型
• type(name, bases, namespace) 动态创建类
• 自定义元类继承自type

元类核心实现

"""
Python元类高级应用示例
展示元类在ORM、API框架和插件系统中的应用
"""

import inspect
from typing import Dict, List, Any, Callable, Optional
from functools import wraps


# ============ 基础元类示例 ============
class AutoRegisterMeta(type):
    """自动注册类到全局注册表的元类"""
    
    registry = {}
    
    def __new__(mcs, name, bases, namespace):
        # 创建类
        cls = super().__new__(mcs, name, bases, namespace)
        
        # 自动注册（排除元类本身）
        if name != 'AutoRegisterMeta':
            mcs.registry[name] = cls
            print(f"[Meta] 注册类: {name}")
        
        return cls


class SingletonMeta(type):
    """单例模式元类"""
    
    _instances = {}
    
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


# ============ ORM框架元类 ============
class Field:
    """ORM字段基类"""
    def __init__(self, field_type: str, primary_key: bool = False, nullable: bool = True):
        self.field_type = field_type
        self.primary_key = primary_key
        self.nullable = nullable
        self.name = None
    
    def __set_name__(self, owner, name):
        self.name = name


class IntegerField(Field):
    def __init__(self, primary_key: bool = False, nullable: bool = True):
        super().__init__("INTEGER", primary_key, nullable)


class StringField(Field):
    def __init__(self, max_length: int = 255, nullable: bool = True):
        super().__init__(f"VARCHAR({max_length})", False, nullable)
        self.max_length = max_length


class ModelMeta(type):
    """ORM模型元类 - 自动处理字段映射"""
    
    def __new__(mcs, name, bases, namespace):
        if name == 'Model':
            return super().__new__(mcs, name, bases, namespace)
        
        # 收集字段定义
        fields = {}
        for key, value in list(namespace.items()):
            if isinstance(value, Field):
                fields[key] = value
        
        # 创建类
        cls = super().__new__(mcs, name, bases, namespace)
        
        # 设置元数据
        cls._fields = fields
        cls._table_name = name.lower()
        
        # 生成SQL创建语句
        cls._create_table_sql = mcs._generate_create_table(name, fields)
        
        print(f"[ORM] 生成模型: {name}, 字段: {list(fields.keys())}")
        return cls
    
    @staticmethod
    def _generate_create_table(name: str, fields: Dict[str, Field]) -> str:
        """生成CREATE TABLE语句"""
        columns = []
        primary_keys = []
        
        for field_name, field in fields.items():
            column_def = f"{field_name} {field.field_type}"
            if not field.nullable:
                column_def += " NOT NULL"
            if field.primary_key:
                primary_keys.append(field_name)
            columns.append(column_def)
        
        if primary_keys:
            columns.append(f"PRIMARY KEY ({', '.join(primary_keys)})")
        
        return f"CREATE TABLE {name.lower()} ({', '.join(columns)});"


class Model(metaclass=ModelMeta):
    """ORM模型基类"""
    
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            if key in self._fields:
                setattr(self, key, value)
    
    def save(self):
        """保存到数据库（模拟）"""
        values = {k: getattr(self, k, None) for k in self._fields.keys()}
        print(f"[ORM] 保存 {self.__class__.__name__}: {values}")
        return self
    
    @classmethod
    def create_table(cls):
        """创建数据表"""
        print(f"[ORM] 执行SQL: {cls._create_table_sql}")


# ============ API框架元类 ============
class Route:
    """路由装饰器"""
    def __init__(self, path: str, method: str = "GET"):
        self.path = path
        self.method = method
    
    def __call__(self, func):
        func._route = {"path": self.path, "method": self.method}
        return func


class APIMeta(type):
    """API控制器元类 - 自动注册路由"""
    
    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        
        if name == 'APIController':
            return cls
        
        # 收集路由
        routes = {}
        for attr_name, attr_value in namespace.items():
            if callable(attr_value) and hasattr(attr_value, '_route'):
                route_info = attr_value._route
                routes[route_info['path']] = {
                    'method': route_info['method'],
                    'handler': attr_value
                }
        
        cls._routes = routes
        cls._base_path = f"/api/{name.lower().replace('controller', '')}"
        
        print(f"[API] 注册控制器: {name}, 路由: {list(routes.keys())}")
        return cls


class APIController(metaclass=APIMeta):
    """API控制器基类"""
    
    @classmethod
    def dispatch(cls, path: str, method: str, **kwargs):
        """分发请求"""
        full_path = cls._base_path + path
        
        for route_path, route_info in cls._routes.items():
            if route_path == path and route_info['method'] == method:
                handler = route_info['handler']
                return handler(cls(), **kwargs)
        
        return {"error": "Not Found", "status": 404}


# ============ 插件系统元类 ============
class PluginMeta(type):
    """插件系统元类 - 自动发现和管理插件"""
    
    plugins = {}
    hooks = {}
    
    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        
        if name == 'Plugin':
            return cls
        
        # 注册插件
        plugin_name = getattr(cls, 'name', name.lower())
        mcs.plugins[plugin_name] = cls
        
        # 收集钩子方法
        hooks = {}
        for attr_name, attr_value in namespace.items():
            if callable(attr_value) and hasattr(attr_value, '_hook'):
                hook_name = attr_value._hook
                if hook_name not in mcs.hooks:
                    mcs.hooks[hook_name] = []
                mcs.hooks[hook_name].append((plugin_name, attr_value))
        
        print(f"[Plugin] 注册插件: {plugin_name}")
        return cls
    
    @classmethod
    def execute_hook(mcs, hook_name: str, *args, **kwargs):
        """执行钩子"""
        results = []
        if hook_name in mcs.hooks:
            for plugin_name, hook_func in mcs.hooks[hook_name]:
                try:
                    result = hook_func(mcs.plugins[plugin_name](), *args, **kwargs)
                    results.append((plugin_name, result))
                except Exception as e:
                    print(f"[Plugin] 钩子执行错误 {plugin_name}.{hook_name}: {e}")
        return results


def hook(hook_name: str):
    """钩子装饰器"""
    def decorator(func):
        func._hook = hook_name
        return func
    return decorator


class Plugin(metaclass=PluginMeta):
    """插件基类"""
    name = "base_plugin"
    version = "1.0.0"
    
    def initialize(self):
        """初始化插件"""
        print(f"[Plugin] 初始化 {self.name}")


# ============ 实际应用示例 ============

# 1. 使用ORM元类
class User(Model):
    """用户模型"""
    id = IntegerField(primary_key=True)
    username = StringField(max_length=50, nullable=False)
    email = StringField(max_length=100)
    age = IntegerField()


class Post(Model):
    """文章模型"""
    id = IntegerField(primary_key=True)
    title = StringField(max_length=200)
    content = StringField(max_length=5000)
    user_id = IntegerField()


# 2. 使用API元类
class UserController(APIController):
    """用户API控制器"""
    
    @Route("/list", "GET")
    def list_users(self):
        return {"users": [], "count": 0}
    
    @Route("/create", "POST")
    def create_user(self, **data):
        return {"message": "User created", "data": data}
    
    @Route("/<id>", "GET")
    def get_user(self, id=None):
        return {"user_id": id, "name": "Test User"}


# 3. 使用插件元类
class LoggingPlugin(Plugin):
    """日志插件"""
    name = "logging"
    
    @hook("before_request")
    def log_request(self, request):
        print(f"[Log] Request: {request}")
        return "logged"
    
    @hook("after_response")
    def log_response(self, response):
        print(f"[Log] Response: {response}")
        return "logged"


class AuthPlugin(Plugin):
    """认证插件"""
    name = "auth"
    
    @hook("before_request")
    def check_auth(self, request):
        print(f"[Auth] Checking: {request}")
        return "authenticated"


def main():
    """主函数 - 演示元类应用"""
    print("\n" + "="*60)
    print("Python元类高级应用演示")
    print("="*60)
    
    # 演示ORM
    print("\n【ORM框架演示】")
    User.create_table()
    Post.create_table()
    
    user = User(id=1, username="alice", email="alice@example.com", age=25)
    user.save()
    
    # 演示API路由
    print("\n【API框架演示】")
    result1 = UserController.dispatch("/list", "GET")
    print(f"路由响应: {result1}")
    
    result2 = UserController.dispatch("/create", "POST", username="bob")
    print(f"路由响应: {result2}")
    
    # 演示插件系统
    print("\n【插件系统演示】")
    print(f"已注册插件: {list(PluginMeta.plugins.keys())}")
    print(f"可用钩子: {list(PluginMeta.hooks.keys())}")
    
    # 执行钩子
    results = PluginMeta.execute_hook("before_request", {"url": "/api/users"})
    print(f"钩子执行结果: {results}")
    
    # 演示单例
    print("\n【单例模式演示】")
    
    class Database(metaclass=SingletonMeta):
        def __init__(self):
            self.connection = "connected"
    
    db1 = Database()
    db2 = Database()
    print(f"db1 is db2: {db1 is db2}")
    
    print("\n" + "="*60)
    print("元类应用总结")
    print("="*60)
    print("1. ORM框架: 自动映射类属性到数据库字段")
    print("2. API框架: 自动注册路由和处理请求")
    print("3. 插件系统: 自动发现和执行钩子函数")
    print("4. 单例模式: 确保类只有一个实例")
    print("5. 自动注册: 类创建时自动加入注册表")
    print("="*60)


if __name__ == "__main__":
    main()

元类工作流程图

元类应用场景选择

关键要点

1. 元类是类的模板：控制类的创建过程，而非实例的创建
2. new vs init：__new__创建类对象，__init__初始化类对象
3. 应用场景：ORM、API框架、插件系统、单例模式、自动注册
4. 注意事项：元类会增加代码复杂度，仅在必要时使用

通过合理使用元类，可以构建出强大且灵活的框架系统，大幅提升开发效率。