Python元类编程与ORM框架实现

引言

去年我接手一个内部数据平台的重构，需求很简单：把散落在十几个脚本里的 SQL 拼接逻辑收敛成统一的模型层。团队里有人提议直接上 SQLAlchemy，但甲方要求"零第三方 ORM 依赖"，因为部署环境是信创体系下的私有云，依赖审计极其严格。

于是我们花了三周手写了一个微型 ORM，核心只有 800 行代码。写完之后，团队里一位工作五年的同事说了句话让我印象深刻："原来 Django ORM 那些 CharField(max_length=32) 的魔法，本质就是元类在类创建的那一刻偷梁换柱。"

这句话点破了元类编程的核心价值：元类让你能在类"出生"的瞬间，对它进行改造。这不是语法糖，而是 Python 对象模型最深的一层扩展点。Django ORM、SQLAlchemy 的 declarative base、Pydantic 的模型校验、FastAPI 的依赖注入，全都建立在这个机制之上。

这篇文章会从 Python 对象模型的源码级原理讲起，手写一个可运行的微型 ORM，再对比业界主流框架的实现差异，最后给出我在生产环境中踩过的坑。目标读者是已经写过三五年 Python、但还没真正吃透元类的人。

---

核心概念：从"类也是对象"说起

生活化类比：餐厅的菜单与后厨

想象你走进一家餐厅。你手里拿到的菜单是静态的——上面印着菜名、价格、配料。但后厨是动态的：你点单的瞬间，厨师才开始组合食材。

在 Python 里：

• 普通实例是你点的那盘菜——具体、可食用、一次性的。

• 类是菜单——它规定了"这类菜"应该长什么样，但它本身也是一份可以被修改的印刷品（因为类也是对象）。

• 元类是印刷厂——它决定了菜单怎么印、印几页、用什么字体。当你说"给我印一份新菜单"（定义一个新类）时，印刷厂（元类）的流水线就会启动。

关键洞察：菜单也是纸，纸也能被印刷厂处理。这就是 type 既是所有类的类型、又是 type 自身的类型的原因。

技术定义

在 CPython 中，一切皆对象，每个对象都有 ob_type 指针指向它的类型。类的创建过程遵循这条链路：

graph TD A["用户写 class Foo: pass"] --> B["Python 编译器生成 BUILD_CLASS 字节码"] B --> C["调用 type.__call__(metaclass, name, bases, namespace)"] C --> D["metaclass.__new__ 创建类对象"] D --> E["metaclass.__init__ 初始化类对象"] E --> F["返回类对象 Foo"] F --> G["Foo.__class__ == metaclass"] G --> H["metaclass.__class__ == type"]

默认情况下，metaclass 就是 type。当你在类定义里写 class Foo(metaclass=MyMeta)，Python 就会用 MyMeta 走上面这条流水线。

一个容易被忽略的细节：元类的选择遵循"最派生原则"。如果父类的元类是 MetaA，子类显式指定 MetaB，而 MetaB 不是 MetaA 的子类，就会抛 TypeError: metaclass conflict。这是元类继承链的硬约束，Django 的 ModelBase 就踩在这个点上——它必须继承 type，同时 Model 的子类如果想混入自定义元类，必须让自定义元类继承 ModelBase。

---

源码/原理深度分析

type.__call__ 的三段式

CPython 的 type_call（位于 Objects/typeobject.c）逻辑非常清晰，伪代码如下：

static PyObject *
type_call(PyTypeObject *type, PyObject *args, PyObject *kwds)
{
    // 1. 调用 tp_new 创建实例（对元类来说，实例就是类对象）
    obj = type->tp_new(type, args, kwds);
    // 2. 如果 tp_init 存在，调用它
    if (type->tp_init != NULL) {
        type->tp_init(obj, args, kwds);
    }
    // 3. 返回
    return obj;
}

对元类而言，tp_new 就是 metaclass.__new__，tp_init 就是 metaclass.__init__。__new__ 负责"造出来"，__init__ 负责"初始化"。这跟普通实例化完全同构，只是作用对象从"实例"变成了"类"。

这里有个关键点：type.__new__(mcls, name, bases, namespace) 会做大量工作——解析 __slots__、计算 MRO（__mro__）、处理 __set_name__ 描述符回调、合并 __dict__。所以自定义元类时，__new__ 里必须调用 super().__new__，且返回的对象必须是类，否则整个类创建会失败。

__prepare__：命名空间的"预分配"

Python 3 引入了一个容易被忽略的钩子 __prepare__。它在类体代码执行之前被调用，返回一个映射对象作为类体的局部命名空间。

class Meta(type):
    @classmethod
    def __prepare__(mcls, name, bases, **kwds):
        print(f"preparing namespace for {name}")
        return {}  # 必须是可变映射

这个钩子的价值在于：你可以返回一个有序字典（Python 3.6+ 默认 dict 已有序，但历史上有 OrderedDict 方案）或带副作用的自定义映射，从而在类体执行时就捕获字段定义顺序。Django ORM 的字段顺序、SQLAlchemy 的列顺序，都依赖这个机制。

描述符：字段访问的"门卫"

元类负责"类创建时"的改造，但实例属性访问的魔法靠的是描述符协议。一个类如果实现了 __get__/__set__/__delete__，它的实例就成为描述符。

class Field:
    def __init__(self, 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):
        instance.__dict__[self.name] = value

当 User.name 被访问时，Python 的 object.__getattribute__（Objects/object.c 里的 PyObject_GenericGetAttr）会先查 type(instance).__mro__ 里是否有同名的数据描述符（同时实现 __get__ 和 __set__），有则优先调用描述符。这是 ORM 能实现"字段级钩子"的底层保障。

元类 + 描述符 = ORM 的两大支柱

把上面两块拼起来，ORM 的骨架就清晰了：

graph LR A["class User(Model)"] --> B["ModelBase.__new__"] B --> C["遍历 namespace 中的 Field 实例"] C --> D["收集到 _fields 字典"] D --> E["调用 type.__new__ 生成类"] E --> F["实例化 User()"] F --> G["访问 user.name 触发 Field.__get__"] G --> H["从 instance.__dict__ 取值"]

元类在"类创建期"做静态收集，描述符在"实例访问期"做动态拦截。前者是编译期元编程，后者是运行期元编程。两者分工明确，缺一不可。

---

实战代码

示例一：最小可用的元类 ORM

这个版本能跑通 CREATE TABLE、INSERT、SELECT，代码控制在 150 行内，注释详细到可以当教学材料。

import sqlite3
from typing import Any, Dict, List, Optional


class Field:
    """字段描述符：同时承担类型声明和值存取两个职责"""

    def __init__(self, column_type: str, primary_key: bool = False,
                 nullable: bool = True, default: Any = None):
        self.column_type = column_type
        self.primary_key = primary_key
        self.nullable = nullable
        self.default = default
        self.name: Optional[str] = None  # 由元类在 __set_name__ 或 __new__ 中回填

    def __set_name__(self, owner, name):
        # Python 3.6+ 的钩子：描述符被赋值给类属性时自动调用
        # 比在元类里手动回填更优雅，但元类里也能做，二者可互补
        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):
        instance.__dict__[self.name] = value

    def to_sql(self) -> str:
        """生成建表用的列定义"""
        parts = [f'"{self.name}" {self.column_type}']
        if self.primary_key:
            parts.append("PRIMARY KEY")
        if not self.nullable:
            parts.append("NOT NULL")
        return " ".join(parts)


class ModelMeta(type):
    """元类：在类创建时收集所有 Field，生成 _fields 元信息"""

    def __new__(mcls, name, bases, namespace, **kwargs):
        # 1. 先把父类的字段继承下来（支持 ORM 继承）
        fields: Dict[str, Field] = {}
        for base in bases:
            if hasattr(base, "_fields"):
                fields.update(base._fields)

        # 2. 扫描当前类体里新定义的 Field
        for attr_name, attr_value in namespace.items():
            if isinstance(attr_value, Field):
                # __set_name__ 尚未触发（它在 type.__new__ 内部才调用），
                # 这里先手动回填，保证后续逻辑可用
                attr_value.name = attr_name
                fields[attr_name] = attr_value

        # 3. 把 _fields 塞进命名空间，供实例方法使用
        namespace["_fields"] = fields
        namespace["_table_name"] = namespace.get("__tablename__", name.lower())

        # 4. 交给 type 完成真正的类创建
        cls = super().__new__(mcls, name, bases, namespace, **kwargs)
        return cls


class Model(metaclass=ModelMeta):
    """所有模型类的基类"""

    _fields: Dict[str, Field] = {}
    _table_name: str = ""

    # ---- 类方法：DDL ----
    @classmethod
    def create_table(cls, conn: sqlite3.Connection):
        cols = ", ".join(f.to_sql() for f in cls._fields.values())
        sql = f'CREATE TABLE IF NOT EXISTS "{cls._table_name}" ({cols})'
        conn.execute(sql)
        conn.commit()

    # ---- 实例方法：DML ----
    def save(self, conn: sqlite3.Connection):
        names = list(self._fields.keys())
        placeholders = ", ".join("?" for _ in names)
        col_str = ", ".join(f'"{n}"' for n in names)
        values = [getattr(self, n) for n in names]
        sql = f'INSERT INTO "{self._table_name}" ({col_str}) VALUES ({placeholders})'
        conn.execute(sql, values)
        conn.commit()

    @classmethod
    def query(cls, conn: sqlite3.Connection, where: str = "", params: tuple = ()):
        sql = f'SELECT * FROM "{cls._table_name}"'
        if where:
            sql += f" WHERE {where}"
        cursor = conn.execute(sql, params)
        columns = [d[0] for d in cursor.description]
        results = []
        for row in cursor.fetchall():
            obj = cls()
            for col, val in zip(columns, row):
                setattr(obj, col, val)
            results.append(obj)
        return results

    def __repr__(self):
        pairs = ", ".join(f"{k}={getattr(self, k)!r}" for k in self._fields)
        return f"{type(self).__name__}({pairs})"


# ---------------- 使用示例 ----------------
class User(Model):
    __tablename__ = "users"
    id = Field("INTEGER", primary_key=True)
    name = Field("TEXT", nullable=False)
    email = Field("TEXT")


if __name__ == "__main__":
    conn = sqlite3.connect(":memory:")
    User.create_table(conn)

    u = User()
    u.id = 1
    u.name = "架构师老王"
    u.email = "laowang@example.com"
    u.save(conn)

    print(User.query(conn))  # [User(id=1, name='架构师老王', email='laowang@example.com')]
    print("字段元信息:", list(User._fields.keys()))  # ['id', 'name', 'email']

几个设计取舍值得说明：

1. 为什么用 __set_name__ 还要在元类里手动回填？ 因为 __set_name__ 是在 type.__new__ 内部触发的，而元类 __new__ 中扫描 namespace 的时机在 type.__new__ 之前。手动回填保证 fields 字典里的 Field.name 立即可用。这是 Django 早期版本的做法，后来改用 contribute_to_class 钩子更优雅。

1. _fields 为什么放命名空间而不是类属性？ 因为 type.__new__ 返回的类对象才是最终产物，元类 __new__ 里对 namespace 的修改会进入类 __dict__。这是最干净的注入方式。

1. query 为什么用 setattr 而不是直接塞 __dict__？ 因为 setattr 会走描述符协议，未来如果要加类型校验、字段加密，扩展点已经就位。

示例二：带查询构建器与类型校验的进阶版

生产 ORM 不可能只有 save/query。这个版本加入了 QuerySet 链式调用、字段类型校验、以及 __init__ 自动赋值。

import sqlite3
from typing import Any, Dict, List, Optional, Tuple


class ValidationError(Exception):
    pass


class Field:
    def __init__(self, column_type: str, primary_key: bool = False,
                 nullable: bool = True, validator=None):
        self.column_type = column_type
        self.primary_key = primary_key
        self.nullable = nullable
        self.validator = validator
        self.name: Optional[str] = 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 value is None and not self.nullable:
            raise ValidationError(f"字段 {self.name} 不允许为空")
        if value is not None and self.validator:
            self.validator(value)
        instance.__dict__[self.name] = value

    def to_sql(self) -> str:
        parts = [f'"{self.name}" {self.column_type}']
        if self.primary_key:
            parts.append("PRIMARY KEY")
        if not self.nullable:
            parts.append("NOT NULL")
        return " ".join(parts)


class QuerySet:
    """惰性查询集：链式调用直到 __iter__ 才真正执行 SQL"""

    def __init__(self, model_cls, conn: sqlite3.Connection):
        self.model_cls = model_cls
        self.conn = conn
        self._where: List[str] = []
        self._params: List[Any] = []
        self._order: List[str] = []
        self._limit: Optional[int] = None

    def filter(self, **kwargs) -> "QuerySet":
        for k, v in kwargs.items():
            if k not in self.model_cls._fields:
                raise ValueError(f"未知字段 {k}")
            self._where.append(f'"{k}" = ?')
            self._params.append(v)
        return self

    def order_by(self, *cols: str) -> "QuerySet":
        self._order.extend(cols)
        return self

    def limit(self, n: int) -> "QuerySet":
        self._limit = n
        return self

    def _build_sql(self) -> Tuple[str, tuple]:
        sql = f'SELECT * FROM "{self.model_cls._table_name}"'
        if self._where:
            sql += " WHERE " + " AND ".join(self._where)
        if self._order:
            sql += " ORDER BY " + ", ".join(self._order)
        if self._limit:
            sql += f" LIMIT {self._limit}"
        return sql, tuple(self._params)

    def __iter__(self):
        sql, params = self._build_sql()
        cursor = self.conn.execute(sql, params)
        columns = [d[0] for d in cursor.description]
        for row in cursor.fetchall():
            obj = self.model_cls.__new__(self.model_cls)  # 绕过 __init__ 直接造对象
            for col, val in zip(columns, row):
                obj.__dict__[col] = val  # 直接写 __dict__，避免校验开销
            yield obj

    def first(self):
        return next(iter(self.limit(1)), None)

    def all(self) -> List:
        return list(self)


class ModelMeta(type):
    def __new__(mcls, name, bases, namespace, **kwargs):
        fields: Dict[str, Field] = {}
        for base in bases:
            fields.update(getattr(base, "_fields", {}))

        for attr_name, attr_value in namespace.items():
            if isinstance(attr_value, Field):
                attr_value.name = attr_name
                fields[attr_name] = attr_value

        namespace["_fields"] = fields
        namespace["_table_name"] = namespace.get("__tablename__", name.lower())
        return super().__new__(mcls, name, bases, namespace, **kwargs)


class Model(metaclass=ModelMeta):
    _fields: Dict[str, Field] = {}
    _table_name: str = ""
    _conn: Optional[sqlite3.Connection] = None

    @classmethod
    def bind(cls, conn: sqlite3.Connection):
        """绑定数据库连接，全局共享"""
        cls._conn = conn

    @classmethod
    def create_table(cls, conn: sqlite3.Connection = None):
        conn = conn or cls._conn
        cols = ", ".join(f.to_sql() for f in cls._fields.values())
        conn.execute(f'CREATE TABLE IF NOT EXISTS "{cls._table_name}" ({cols})')
        conn.commit()

    def __init__(self, **kwargs):
        for k in self._fields:
            if k in kwargs:
                setattr(self, k, kwargs[k])

    def save(self):
        names = list(self._fields.keys())
        placeholders = ", ".join("?" for _ in names)
        col_str = ", ".join(f'"{n}"' for n in names)
        values = [getattr(self, n) for n in names]
        self._conn.execute(
            f'INSERT INTO "{self._table_name}" ({col_str}) VALUES ({placeholders})',
            values,
        )
        self._conn.commit()

    @classmethod
    def objects(cls) -> QuerySet:
        return QuerySet(cls, cls._conn)

    def __repr__(self):
        pairs = ", ".join(f"{k}={getattr(self, k)!r}" for k in self._fields)
        return f"{type(self).__name__}({pairs})"


# ---------------- 使用示例 ----------------
def validate_email(v: str):
    if "@" not in v:
        raise ValidationError(f"非法邮箱: {v}")


class User(Model):
    __tablename__ = "users"
    id = Field("INTEGER", primary_key=True)
    name = Field("TEXT", nullable=False)
    email = Field("TEXT", validator=validate_email)


if __name__ == "__main__":
    conn = sqlite3.connect(":memory:")
    User.bind(conn)
    User.create_table()

    User(name="老王", email="laowang@example.com").save()
    User(name="小李", email="xiaoli@example.com").save()
    User(name="小张", email="zhang@example.com").save()

    # 链式查询：惰性执行
    result = User.objects().filter(name="老王").filter(email="laowang@example.com").all()
    print(result)  # [User(id=1, name='老王', email='laowang@example.com')]

    # 类型校验：非法邮箱被拦截
    try:
        u = User()
        u.email = "not-an-email"
    except ValidationError as e:
        print("校验失败:", e)  # 校验失败: 非法邮箱: not-an-email

关键设计点：

• QuerySet 的惰性求值让链式 API 可以任意组合，直到 __iter__ 才拼 SQL。这是 Django ORM 的核心体验，也是性能优化的基础（可以在 __iter__ 里做 SQL 缓存、分页、连接池选择）。

• __new__ 绕过 __init__ 造对象，直接写 __dict__，避免校验开销——这是 ORM 从数据库读数据的标准优化手法。Django 的 from_db 类方法就是干这个的。

• bind 用类属性共享连接，简化了示例。生产中应该用连接池 + 上下文管理，这里为了聚焦元类主题做了简化。

示例三：元类实现依赖注入容器（FastAPI 风格）

元类不只用于 ORM。FastAPI 的 Depends 机制、Spring 的 @Autowired，本质都是"在类创建时收集依赖声明，在实例化时注入"。这个例子展示元类在 DI 场景的用法。

import inspect
from typing import Any, Callable, Dict, List, Type


class Injectable:
    """标记类可被依赖注入容器管理"""


class Container:
    """极简 DI 容器：支持单例与瞬时两种生命周期"""

    _registry: Dict[str, Dict[str, Any]] = {}

    @classmethod
    def register(cls, name: str, factory: Callable, singleton: bool = True):
        cls._registry[name] = {"factory": factory, "singleton": singleton, "instance": None}

    @classmethod
    def resolve(cls, name: str) -> Any:
        entry = cls._registry.get(name)
        if entry is None:
            raise LookupError(f"未注册的依赖: {name}")
        if entry["singleton"]:
            if entry["instance"] is None:
                entry["instance"] = entry["factory"]()
            return entry["instance"]
        return entry["factory"]()


class InjectMeta(type):
    """
    元类职责：
    1. 扫描类体中的 __inject__ 声明（字符串 -> 类型映射）
    2. 包装 __init__，在实例化时自动从容器取值注入
    3. 记录到容器，支持按名解析
    """

    def __new__(mcls, name, bases, namespace, **kwargs):
        inject_spec: Dict[str, Type] = namespace.get("__inject__", {})

        original_init = namespace.get("__init__")

        def wrapped_init(self, *args, **kw):
            # 先注入依赖
            for attr_name, dep_name in inject_spec.items():
                setattr(self, attr_name, Container.resolve(dep_name))
            # 再走用户自己的 __init__（如果有）
            if original_init:
                original_init(self, *args, **kw)

        # 保留原 __init__ 的签名信息，方便 IDE 和文档工具识别
        if original_init:
            wrapped_init.__wrapped__ = original_init
            try:
                wrapped_init.__signature__ = inspect.signature(original_init)
            except (ValueError, TypeError):
                pass

        namespace["__init__"] = wrapped_init
        cls = super().__new__(mcls, name, bases, namespace, **kwargs)

        # 自动注册到容器（按类名注册）
        Container.register(name, cls, singleton=False)
        return cls


# ---------------- 使用示例 ----------------
class Database:
    def __init__(self):
        self.connected = True
        print("[Database] 连接已建立")

    def query(self, sql: str):
        return f"执行: {sql}"


class Cache:
    def __init__(self):
        print("[Cache] 缓存客户端初始化")

    def get(self, key: str):
        return f"cache[{key}]"


# 先注册底层依赖（单例）
Container.register("db", Database, singleton=True)
Container.register("cache", Cache, singleton=True)


class UserService(metaclass=InjectMeta):
    __inject__ = {
        "db": "db",
        "cache": "cache",
    }

    def __init__(self, prefix: str = "svc"):
        self.prefix = prefix

    def get_user(self, uid: int):
        cached = self.cache.get(f"user:{uid}")
        if cached:
            return cached
        return self.db.query(f"SELECT * FROM users WHERE id={uid}")


if __name__ == "__main__":
    svc1 = UserService(prefix="order")
    svc2 = UserService(prefix="pay")
    print(svc1.get_user(1))
    print(svc2.get_user(2))
    # 验证单例：两次注入的是同一个 Database 实例
    print("单例验证:", svc1.db is svc2.db)  # True

为什么用元类而不是装饰器？

装饰器也能包装 __init__，但元类有两个优势：

1. 继承自动生效：子类会自然继承元类，__inject__ 的收集逻辑无需重复声明。装饰器方案下，每个子类都得手动加 @inject。
2. 可以在类创建时做全局注册：Container.register(name, cls) 这一步在元类里执行，类一被定义就进入容器。装饰器要额外调用一次注册函数。

代价是元类的"侵入性"——一旦用了元类，这个类的所有子类都被绑定。这也是为什么 FastAPI 选择用函数签名 + 装饰器而非元类来实现 Depends：它需要更灵活的粒度控制。

---

方案对比：元类 ORM vs 装饰器 ORM vs 声明式 ORM

维度	元类方案（Django/SQLAlchemy Declarative）	装饰器方案（dataclass + 手写映射）	显式注册方案（Peewee 早期风格）
字段声明方式	name = CharField(...)	name: str + @field	name = Column(String) + register()
类创建期改造能力	强（可改 MRO、注入方法）	弱（只能包装类，不能改类体）	无
继承支持	天然支持（元类沿继承链传递）	需手动处理	需手动注册
可读性	中（魔法多）	高	高
调试难度	高（栈深）	低	低
典型代表	Django ORM, SQLAlchemy	dataclasses, Pydantic	Peewee, Tortoise

几个业界实现的细节点：

• Django ORM 的 ModelBase.__new__ 里有约 200 行逻辑，处理继承、抽象基类、多表继承、代理模型、字段冲突检测。它还会调用每个 Field.contribute_to_class(cls, name)，这是比 __set_name__ 更早的钩子（Django 1.x 时代 __set_name__ 还不存在）。

• SQLAlchemy 的 declarative base 用 _as_declarative 函数在 DeclarativeMeta.__init__ 里扫描 cls.__dict__，把 Column 对象搬进 cls.__table__。它的 __init__ 比 __new__ 用得更多，因为需要类对象已存在才能操作 __table__。

• Pydantic v2 已经放弃了元类，改用 __init_subclass__ + Rust 实现的 ModelMetaclass 混合方案。__init_subclass__（PEP 487）是 Python 3.6 引入的轻量钩子，能在子类创建时被父类感知，比元类侵入性小得多。如果你的需求只是"子类定义时做点事"，优先用 __init_subclass__。

# __init_subclass__ 的等价写法，比元类轻量得多
class Model:
    _fields = {}
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._fields = {
            k: v for k, v in cls.__dict__.items()
            if isinstance(v, Field)
        }

---

最佳实践与避坑指南

坑一：元类冲突（metaclass conflict）

class MetaA(type): pass
class MetaB(type): pass
class A(metaclass=MetaA): pass
class B(A, metaclass=MetaB): pass  # TypeError!

原因：Python 要求最终元类必须是所有父类元类的子类。MetaB 不是 MetaA 的子类，无法满足。

解法：要么让 MetaB(MetaA) 继承，要么用 class B(A, metaclass=type('MetaAB', (MetaA, MetaB), {})) 动态合成。Django 的 ModelBase 和 ABCMeta 混用时，官方推荐 class MyModel(Model, metaclass=ABCMeta) 会失败，必须写 class MyModel(Model, metaclass=type('Meta', (ModelBase, ABCMeta), {}))。这是无数人踩过的经典坑。

坑二：__new__ 里忘记调用 super().__new__

class BadMeta(type):
    def __new__(mcls, name, bases, ns):
        ns["foo"] = 1
        # 忘了 return super().__new__(...)

结果是 None，后续所有类操作全部 AttributeError。元类 __new__ 必须返回一个类对象。同理，__init__ 里也必须调 super().__init__，否则 __set_name__ 等内部钩子不会触发。

坑三：描述符的 __get__ 里用 getattr(instance, self.name) 造成无限递归

def __get__(self, instance, owner):
    return getattr(instance, self.name)  # 死循环！

因为 getattr 会再次触发描述符。必须直接访问 instance.__dict__，或者用 object.__getattribute__(instance, self.name)（后者在 __dict__ 没有时仍可能递归，需谨慎）。

坑四：元类影响 pickle 和 copy

元类生成的类在跨进程序列化时，pickle 要求类能被 import 路径找到。动态创建的类（如 type("X", (), {})）无法 pickle。Django 的 Model 通过 __reduce__ 自定义序列化绕过了这个问题。如果你的 ORM 需要支持缓存序列化，务必实现 __reduce__ 或 __getstate__。

坑五：_fields 字典的继承污染

class Base(Model):
    id = Field("INTEGER")
class Child(Base):
    name = Field("TEXT")
# Child._fields 如果直接引用 Base._fields，会互相污染

必须用 dict(getattr(base, "_fields", {})) 做浅拷贝，示例一里已经处理。SQLAlchemy 用 _decl_class_registry 全局注册表 + 每次新建 Table 对象来规避。

最佳实践清单

1. 能用 __init_subclass__ 就别用元类。元类是"核武器"，只在需要改 MRO、改 __prepare__、控制类创建全过程时才用。
2. 元类只做"收集"，别做"执行"。__new__ 里只应扫描和注入元信息，业务逻辑放到 contribute_to_class 之类的钩子里，便于测试。
3. 字段描述符必须实现 __set_name__，让字段自己知道自己叫什么，比在元类里手动回填更健壮。
4. 给元类生成的类补上 __qualname__ 和 __module__，否则日志和调试会显示 type 而非真实类名。
5. 单元测试要覆盖"类定义本身"。元类的 bug 往往在 import 时就暴露，写一个 test_models_importable() 能挡住 80% 的低级错误。

---

总结

回到开头那个 800 行的微型 ORM。写完之后我最大的感触是：元类不是"高级语法"，而是 Python 对象模型的必然产物。因为类也是对象，所以类也需要一个"类型"来创建它，这个类型就是元类。理解这一点，Django ORM 的 CharField、Pydantic 的 BaseModel、FastAPI 的依赖注入，就都从"魔法"变成了"可推导的工程决策"。

几个延伸思考方向，留给读者：

• __init_subclass__ 与元类的边界在哪里？ PEP 487 的设计初衷就是"用 90% 的场景替代元类"。什么时候必须用元类？答案是：当你需要控制 __prepare__（命名空间类型）、需要修改 MRO、或者需要在 type.__new__ 之前干预时。

• 异步 ORM 的元类有什么不同？ SQLAlchemy 2.0 的 async 支持、Tortoise ORM 的 Model，元类职责基本不变，但 QuerySet 的求值时机从 __iter__ 挪到了 await，需要额外处理 __await__ 协议。

• Rust/Pydantic v2 的启示：Pydantic v2 把校验逻辑下沉到 Rust 扩展，元类只做 schema 收集。这提示我们：元类适合做"编译期"的静态分析，运行期热路径应该交给更底层的实现。

最后一句经验：元类写得越少，代码越健康。我见过太多项目把业务逻辑塞进 __new__，结果一个 import 错误能花两小时定位。把元类当成"类定义的编译器前端"，只做语法分析和符号收集，剩下的交给描述符和普通方法——这才是它在生产环境里最舒服的姿势。