元类（Metaclass）：类背后的类
在 Python 中，一切皆对象，类也不例外。类是元类的实例，正如对象是类的实例。默认情况下，所有类的元类都是 type。元类控制类的创建过程，是框架设计中最强大的元编程工具。
理解元类需要首先理解类的创建过程：type(name, bases, namespace) 是创建类的实际函数，其中 name 是类名，bases 是基类元组，namespace 是类属性字典。元类通过重写 __new__ 和 __init__ 方法，可以在类创建时注入行为。
# 基础元类示例
class MetaLogger(type):
"""为所有方法添加日志记录的元类"""
def __new__(mcs, name, bases, namespace):
"""控制类的创建——在类对象实际创建之前调用"""
print(f"[Meta] Creating class: {name}")
# 遍历命名空间，为每个函数添加日志包装
for attr_name, attr_value in namespace.items():
if callable(attr_value) and not attr_name.startswith("__"):
namespace[attr_name] = mcs._add_logging(attr_value, name)
# 调用父类的 __new__ 实际创建类对象
cls = super().__new__(mcs, name, bases, namespace)
return cls
def __init__(cls, name, bases, namespace):
"""初始化新创建的类"""
print(f"[Meta] Initializing class: {name}")
super().__init__(name, bases, namespace)
# 添加类级别的属性
cls._created_at = __import__('time').time()
@staticmethod
def _add_logging(func, class_name):
"""为函数添加日志包装器"""
def wrapper(*args, **kwargs):
print(f"[LOG] {class_name}.{func.__name__} called")
return func(*args, **kwargs)
wrapper.__name__ = func.__name__
return wrapper
# 使用元类
class MyService(metaclass=MetaLogger):
"""使用 MetaLogger 元类的服务类"""
def do_work(self):
return "working"
def process_data(self, data):
return f"processed: {data}"
# 实例化并测试
service = MyService()
print(service.do_work()) # [LOG] MyService.do_work called
print(service.process_data("test")) # [LOG] MyService.process_data called
元类的 __new__ 和 __init__ 在类定义时执行，而 __call__ 在类被实例化时执行。这种三阶段生命周期使得元类可以在类创建、初始化、实例化三个时间点注入逻辑。
class SingletonMeta(type):
"""单例模式元类实现"""
_instances: dict = {}
def __call__(cls, *args, **kwargs):
"""控制类的实例化过程"""
if cls not in cls._instances:
# 首次实例化，创建并缓存实例
instance = super().__call__(*args, **kwargs)
cls._instances[cls] = instance
print(f"[Singleton] Created instance of {cls.__name__}")
else:
print(f"[Singleton] Reusing instance of {cls.__name__}")
return cls._instances[cls]
class DatabaseConnection(metaclass=SingletonMeta):
"""数据库连接单例"""
def __init__(self, dsn: str = None):
# 注意：多次调用 __init__ 会重复执行！
self.dsn = dsn
self.connection = None
print(f"DatabaseConnection initialized with DSN: {dsn}")
def connect(self):
if not self.connection:
self.connection = f"connection_to_{self.dsn}"
return self.connection
# 测试单例行为
db1 = DatabaseConnection("postgresql://localhost")
db2 = DatabaseConnection("mysql://remote")
print(f"db1 is db2: {db1 is db2}") # True
print(f"db1.dsn: {db1.dsn}") # postgresql://localhost
print(f"db2.dsn: {db2.dsn}") # postgresql://localhost (未改变！)
# 改进版：带参数验证的单例
class StrictSingletonMeta(type):
"""严格的单例元类，拒绝重复初始化"""
_instances: dict = {}
_initialized: set = set()
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
instance = cls.__new__(cls, *args, **kwargs)
cls._instances[cls] = instance
cls._initialized.add(cls)
instance.__init__(*args, **kwargs)
else:
if args or kwargs:
raise RuntimeError(
f"Singleton {cls.__name__} already initialized, "
"cannot change parameters"
)
return cls._instances[cls]
重点提示：
- 元类的
__new__必须返回类对象，通常调用super().__new__() - 继承元类的子类会自动使用该元类，除非显式指定其他元类
- 单例模式在 Python 中可用模块级变量或
functools.lru_cache简化实现 - 元类冲突：多重继承时若父类有不同元类，Python 会尝试创建派生元类
描述符协议：属性控制的基石
描述符（Descriptor）是 Python 实现属性访问控制的核心机制。property、staticmethod、classmethod 都是基于描述符协议实现的。描述符协议包含 __get__、__set__、__delete__ 三个方法，定义了属性访问的行为。
Data Descriptor（同时定义了 __get__ 和 __set__）优先级高于实例字典中的同名属性；Non-Data Descriptor（只定义 __get__）优先级低于实例字典。这个区别是方法绑定机制的基础。
# 描述符协议详解
class Validator:
"""数据验证描述符"""
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 # 将在 __set_name__ 中设置
def __set_name__(self, owner, name):
"""描述符被分配给类属性时调用（Python 3.6+）"""
self.name = name
self.storage_name = f"_{owner.__name__}__{name}"
def __get__(self, instance, owner):
"""获取属性值
instance: 实例对象（通过类访问时为 None）
owner: 拥有此描述符的类
"""
if instance is None:
return self # 通过类访问时返回描述符本身
return getattr(instance, self.storage_name, None)
def __set__(self, instance, value):
"""设置属性值，执行验证"""
# 类型检查
if self.type_ is not None and not isinstance(value, self.type_):
raise TypeError(
f"{self.name} must be of type {self.type_.__name__}"
)
# 范围检查
if self.min_value is not None and value < self.min_value:
raise ValueError(
f"{self.name} must be >= {self.min_value}"
)
if self.max_value is not None and value > self.max_value:
raise ValueError(
f"{self.name} must be <= {self.max_value}"
)
setattr(instance, self.storage_name, value)
def __delete__(self, instance):
"""删除属性"""
raise AttributeError(f"Cannot delete attribute {self.name}")
class Person:
"""使用描述符验证属性的类"""
name = Validator(type_=str)
age = Validator(min_value=0, max_value=150, type_=int)
salary = Validator(min_value=0, type_=(int, float))
def __init__(self, name: str, age: int, salary: float):
self.name = name
self.age = age
self.salary = salary
# 测试描述符行为
p = Person("Alice", 30, 50000.0)
print(f"Name: {p.name}, Age: {p.age}, Salary: {p.salary}")
# 以下都会抛出异常
try:
p.age = -5
except ValueError as e:
print(f"Validation error: {e}")
try:
p.name = 123
except TypeError as e:
print(f"Type error: {e}")
# 通过类访问描述符
print(Person.age) # <__main__.Validator object at ...>
# 延迟加载描述符（Lazy Property）
import functools
class LazyProperty:
"""只计算一次的属性描述符"""
def __init__(self, func):
self.func = func
self.name = func.__name__
self.__doc__ = func.__doc__
def __get__(self, instance, owner):
if instance is None:
return self
# 计算值并缓存到实例字典
value = self.func(instance)
setattr(instance, self.name, value)
return value
def __set__(self, instance, value):
"""禁止直接设置"""
raise AttributeError(f"Cannot set {self.name}")
class ExpensiveObject:
"""包含昂贵计算的类"""
def __init__(self, data: list):
self.data = data
@LazyProperty
def sorted_data(self):
"""排序数据（只在首次访问时计算）"""
print("Computing sorted_data...")
return sorted(self.data)
@LazyProperty
def statistics(self):
"""统计数据（只在首次访问时计算）"""
print("Computing statistics...")
return {
"sum": sum(self.data),
"mean": sum(self.data) / len(self.data),
"min": min(self.data),
"max": max(self.data),
}
obj = ExpensiveObject([3, 1, 4, 1, 5, 9, 2, 6])
print("First access to sorted_data:")
print(obj.sorted_data) # 计算并缓存
print("Second access to sorted_data:")
print(obj.sorted_data) # 直接返回缓存值
print("\nAccess to statistics:")
print(obj.statistics) # 触发计算
重点提示：
__set_name__在 Python 3.6+ 引入，使得描述符可以知道自己在类中的名称- Non-Data Descriptor 只定义
__get__，实例字典中的同名属性可以覆盖它 - Python 的方法绑定就是通过 Non-Data Descriptor 实现的：
function.__get__返回 bound method - 描述符协议是 ORM 框架（Django ORM、SQLAlchemy）的核心机制
装饰器设计模式
装饰器是 Python 元编程中最常用的工具，它本质上是一个返回函数的高阶函数。Python 提供了函数装饰器、类装饰器以及装饰器栈（多个装饰器叠加）三种形式，每种都有其特定的应用场景。
import functools
import time
from typing import Callable, Any
# ========== 函数装饰器 ==========
def timing_decorator(func: Callable) -> Callable:
"""记录函数执行时间的装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
start = time.perf_counter()
try:
result = func(*args, **kwargs)
return result
finally:
elapsed = time.perf_counter() - start
print(f"[TIMING] {func.__name__} took {elapsed:.4f}s")
return wrapper
def retry_decorator(max_attempts: int = 3, delay: float = 1.0):
"""带参数的装饰器：重试机制"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts:
raise
print(f"[RETRY] Attempt {attempt} failed: {e}")
time.sleep(delay)
return wrapper
return decorator
# 使用示例
@timing_decorator
@retry_decorator(max_attempts=3)
def unstable_api_call():
"""可能失败的 API 调用"""
if time.time() % 2 > 1.5:
raise ConnectionError("Network error")
return "Success"
# ========== 类装饰器 ==========
def singleton(cls):
"""类装饰器实现单例模式"""
instances = {}
lock = __import__('threading').Lock()
@functools.wraps(cls)
def wrapper(*args, **kwargs):
if cls not in instances:
with lock:
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return wrapper
def auto_properties(*field_names):
"""为类自动添加属性"""
def decorator(cls):
for field_name in field_names:
def make_getter(name):
return lambda self: getattr(self, f"_{name}", None)
def make_setter(name):
return lambda self, value: setattr(self, f"_{name}", value)
setattr(cls, field_name, property(
make_getter(field_name),
make_setter(field_name)
))
return cls
return decorator
@auto_properties("name", "age", "email")
class User:
"""自动拥有 name/age/email 属性的类"""
pass
# ========== 装饰器栈的执行顺序 ==========
def decorator_a(func):
print("Decorator A applied")
def wrapper(*args, **kwargs):
print("A: before")
result = func(*args, **kwargs)
print("A: after")
return result
return wrapper
def decorator_b(func):
print("Decorator B applied")
def wrapper(*args, **kwargs):
print("B: before")
result = func(*args, **kwargs)
print("B: after")
return result
return wrapper
@decorator_a
@decorator_b
def my_function():
print("Function executing")
# 执行顺序：
# 1. decorator_b 应用（打印 "Decorator B applied"）
# 2. decorator_a 应用（打印 "Decorator A applied"）
# 3. 调用时：A:before -> B:before -> function -> B:after -> A:after
my_function()
# ========== 方法装饰器（带 self 的类方法） ==========
def method_decorator(func):
"""适用于方法的装饰器"""
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
print(f"[METHOD] {self.__class__.__name__}.{func.__name__}")
return func(self, *args, **kwargs)
return wrapper
class MyClass:
@method_decorator
def do_something(self):
return "done"
# ========== 类方法装饰器 ==========
def classmethod_decorator(func):
"""适用于类方法的装饰器"""
@functools.wraps(func)
def wrapper(cls, *args, **kwargs):
print(f"[CLASSMETHOD] {cls.__name__}.{func.__name__}")
return func(cls, *args, **kwargs)
return wrapper
# ========== 静态方法装饰器 ==========
def staticmethod_decorator(func):
"""适用于静态方法的装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"[STATIC] {func.__name__}")
return func(*args, **kwargs)
return wrapper
# ========== 通用方法装饰器 ==========
def universal_decorator(func):
"""适用于各种方法的通用装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 判断第一个参数是否是实例或类
if args:
first_arg = args[0]
if isinstance(first_arg, type):
print(f"[CALL] classmethod {func.__name__}")
elif hasattr(first_arg, '__class__'):
print(f"[CALL] method {func.__name__}")
else:
print(f"[CALL] function {func.__name__}")
return func(*args, **kwargs)
return wrapper
重点提示：
- 装饰器栈的执行顺序是从下往上应用，调用时从上往下执行
functools.wraps保留原函数的元数据（__name__, __doc__）- 带参数的装饰器实际上是装饰器工厂，返回真正的装饰器
- 类装饰器在类创建后立即执行，可以修改或替换类对象
__init_subclass__ 与协议注册
__init_subclass__ 是 Python 3.6 引入的类方法，在定义子类时自动调用。它提供了一种比元类更轻量的方式来拦截子类创建，常用于插件注册、抽象基类验证等场景。
# 插件注册系统
class PluginRegistry:
"""插件注册中心"""
_plugins: dict = {}
@classmethod
def register(cls, plugin_class):
"""注册插件类"""
cls._plugins[plugin_class.__name__] = plugin_class
print(f"[Plugin] Registered: {plugin_class.__name__}")
@classmethod
def get_plugin(cls, name: str):
"""获取插件类"""
return cls._plugins.get(name)
@classmethod
def list_plugins(cls):
"""列出所有插件"""
return list(cls._plugins.keys())
class PluginBase:
"""插件基类，自动注册子类"""
plugin_name: str = ""
plugin_version: str = "1.0"
def __init_subclass__(cls, **kwargs):
"""子类定义时自动调用"""
super().__init_subclass__(**kwargs)
# 自动注册非抽象的子类
if not getattr(cls, "__abstract__", False):
PluginRegistry.register(cls)
# 验证必需的属性
if not cls.plugin_name:
cls.plugin_name = cls.__name__
def execute(self, *args, **kwargs):
"""插件执行入口"""
raise NotImplementedError
# 具体插件实现
class EmailPlugin(PluginBase):
"""邮件发送插件"""
plugin_name = "email"
plugin_version = "2.0"
def execute(self, to: str, subject: str, body: str):
print(f"[Email] To: {to}, Subject: {subject}")
return {"status": "sent", "to": to}
class SMSPlugin(PluginBase):
"""短信发送插件"""
plugin_name = "sms"
def execute(self, phone: str, message: str):
print(f"[SMS] To: {phone}, Message: {message}")
return {"status": "sent", "phone": phone}
# 抽象插件（不会被注册）
class AbstractNotificationPlugin(PluginBase):
"""抽象通知插件基类"""
__abstract__ = True
# 测试插件系统
print(f"Registered plugins: {PluginRegistry.list_plugins()}")
email_plugin = PluginRegistry.get_plugin("EmailPlugin")()
result = email_plugin.execute("user@example.com", "Hello", "World")
print(f"Result: {result}")
# 使用 __init_subclass__ 实现序列化注册
import json
from typing import Type, Dict
class Serializable:
"""可序列化基类，自动注册序列化器"""
_serializers: Dict[str, Type] = {}
def __init_subclass__(cls, type_name: str = None, **kwargs):
super().__init_subclass__(**kwargs)
# 使用显式指定的类型名或类名
name = type_name or cls.__name__
cls._type_name = name
Serializable._serializers[name] = cls
print(f"[Serializable] Registered: {name} -> {cls.__name__}")
def to_dict(self) -> dict:
"""转换为字典"""
return {
"__type__": self._type_name,
**self._serialize()
}
def _serialize(self) -> dict:
"""子类实现的序列化逻辑"""
raise NotImplementedError
@classmethod
def from_dict(cls, data: dict):
"""从字典反序列化"""
type_name = data.pop("__type__", None)
if type_name not in cls._serializers:
raise ValueError(f"Unknown type: {type_name}")
target_class = cls._serializers[type_name]
return target_class._deserialize(data)
@classmethod
def _deserialize(cls, data: dict):
"""子类实现的反序列化逻辑"""
raise NotImplementedError
# 具体可序列化类
class User(Serializable, type_name="user"):
"""用户模型"""
def __init__(self, name: str, age: int, email: str):
self.name = name
self.age = age
self.email = email
def _serialize(self) -> dict:
return {
"name": self.name,
"age": self.age,
"email": self.email
}
@classmethod
def _deserialize(cls, data: dict):
return cls(**data)
class Product(Serializable, type_name="product"):
"""产品模型"""
def __init__(self, sku: str, name: str, price: float):
self.sku = sku
self.name = name
self.price = price
def _serialize(self) -> dict:
return {
"sku": self.sku,
"name": self.name,
"price": self.price
}
@classmethod
def _deserialize(cls, data: dict):
return cls(**data)
# 测试序列化
user = User("Alice", 30, "alice@example.com")
user_dict = user.to_dict()
print(f"Serialized: {json.dumps(user_dict, indent=2)}")
restored = Serializable.from_dict(user_dict.copy())
print(f"Restored: {restored.name}, {restored.age}")
重点提示：
__init_subclass__比元类更轻量，不会干扰多重继承的元类解析- 可以通过关键字参数向
__init_subclass__传递配置信息 __init_subclass__只处理直接子类，孙类不会触发父类的此方法- 结合
__set_name__可以实现完整的框架级注册系统
实战：构建 ORM 映射引擎
结合元类和描述符协议，我们可以构建一个简化版的 ORM 引擎。这个引擎将支持模型定义、字段验证、查询构建等核心功能。
# mini_orm.py - 简化版 ORM 引擎
from typing import Type, Dict, Any, List, Optional
import sqlite3
# ========== 字段描述符 ==========
class Field:
"""数据库字段基类"""
def __init__(self, name: str = None, primary_key: bool = False,
nullable: bool = True, default=None):
self.name = name
self.primary_key = primary_key
self.nullable = nullable
self.default = default
self.model = None # 将在 __set_name__ 中设置
def __set_name__(self, owner, name):
self.name = name
self.model = owner
owner._fields[name] = self
def __get__(self, instance, owner):
if instance is None:
return self
return instance._data.get(self.name, self.default)
def __set__(self, instance, value):
instance._data[self.name] = value
instance._dirty.add(self.name)
def get_sql_type(self) -> str:
"""返回 SQL 类型定义"""
raise NotImplementedError
class IntegerField(Field):
def get_sql_type(self) -> str:
return "INTEGER"
class TextField(Field):
def get_sql_type(self) -> str:
return "TEXT"
class RealField(Field):
def get_sql_type(self) -> str:
return "REAL"
class BooleanField(Field):
def get_sql_type(self) -> str:
return "INTEGER"
# ========== 模型元类 ==========
class ModelMeta(type):
"""ORM 模型元类"""
def __new__(mcs, name, bases, namespace):
# 初始化字段字典
namespace.setdefault('_fields', {})
# 创建类
cls = super().__new__(mcs, name, bases, namespace)
# 跳过 Model 基类本身
if name == 'Model':
return cls
# 收集所有字段
for base in bases:
if hasattr(base, '_fields'):
cls._fields.update(base._fields)
# 设置表名
if not hasattr(cls, '_table'):
cls._table = name.lower()
# 查找主键
cls._primary_key = None
for field_name, field in cls._fields.items():
if field.primary_key:
cls._primary_key = field_name
break
# 创建查询构建器
cls.objects = QuerySet(cls)
return cls
# ========== 模型基类 ==========
class Model(metaclass=ModelMeta):
"""ORM 模型基类"""
_table: str = ""
_fields: Dict[str, Field] = {}
_primary_key: Optional[str] = None
_db: Optional[sqlite3.Connection] = None
def __init__(self, **kwargs):
self._data: Dict[str, Any] = {}
self._dirty: set = set()
for key, value in kwargs.items():
if key in self._fields:
setattr(self, key, value)
else:
raise AttributeError(f"Unknown field: {key}")
def __repr__(self):
pk = getattr(self, self._primary_key, None) if self._primary_key else None
return f"<{self.__class__.__name__}(pk={pk})>"
def save(self):
"""保存到数据库"""
if self._db is None:
raise RuntimeError("Database not configured")
if self._dirty or not getattr(self, self._primary_key, None):
self._insert_or_update()
self._dirty.clear()
def _insert_or_update(self):
"""执行插入或更新"""
cursor = self._db.cursor()
if getattr(self, self._primary_key, None):
# UPDATE
fields = [f for f in self._dirty if f != self._primary_key]
if fields:
set_clause = ", ".join(f"{f} = ?" for f in fields)
values = [self._data[f] for f in fields]
values.append(self._data[self._primary_key])
sql = f"UPDATE {self._table} SET {set_clause} WHERE {self._primary_key} = ?"
cursor.execute(sql, values)
else:
# INSERT
fields = list(self._data.keys())
placeholders = ", ".join("?" for _ in fields)
values = [self._data[f] for f in fields]
sql = f"INSERT INTO {self._table} ({', '.join(fields)}) VALUES ({placeholders})"
cursor.execute(sql, values)
# 获取自增主键
if self._primary_key:
self._data[self._primary_key] = cursor.lastrowid
self._db.commit()
def delete(self):
"""从数据库删除"""
if self._db is None:
raise RuntimeError("Database not configured")
if not getattr(self, self._primary_key, None):
raise ValueError("Cannot delete unsaved object")
cursor = self._db.cursor()
sql = f"DELETE FROM {self._table} WHERE {self._primary_key} = ?"
cursor.execute(sql, (self._data[self._primary_key],))
self._db.commit()
@classmethod
def connect(cls, database: str):
"""连接数据库"""
cls._db = sqlite3.connect(database)
cls._db.row_factory = sqlite3.Row
@classmethod
def create_table(cls):
"""创建表结构"""
if cls._db is None:
raise RuntimeError("Database not configured")
fields_def = []
for name, field in cls._fields.items():
sql_type = field.get_sql_type()
constraints = []
if field.primary_key:
constraints.append("PRIMARY KEY")
if not field.nullable and not field.primary_key:
constraints.append("NOT NULL")
field_def = f"{name} {sql_type} {' '.join(constraints)}".strip()
fields_def.append(field_def)
sql = f"CREATE TABLE IF NOT EXISTS {cls._table} ({', '.join(fields_def)})"
cls._db.execute(sql)
cls._db.commit()
# ========== 查询集 ==========
class QuerySet:
"""查询构建器"""
def __init__(self, model_class):
self._model = model_class
self._conditions: List[tuple] = []
self._limit: int = None
self._order_by: str = None
def filter(self, **kwargs) -> 'QuerySet':
"""添加过滤条件"""
new_qs = QuerySet(self._model)
new_qs._conditions = self._conditions.copy()
for key, value in kwargs.items():
new_qs._conditions.append((key, '=', value))
return new_qs
def order_by(self, field: str) -> 'QuerySet':
new_qs = QuerySet(self._model)
new_qs._conditions = self._conditions.copy()
new_qs._order_by = field
return new_qs
def limit(self, n: int) -> 'QuerySet':
new_qs = QuerySet(self._model)
new_qs._conditions = self._conditions.copy()
new_qs._limit = n
return new_qs
def all(self) -> List:
"""执行查询"""
if self._model._db is None:
raise RuntimeError("Database not configured")
sql = f"SELECT * FROM {self._model._table}"
values = []
if self._conditions:
where = " AND ".join(f"{k} {op} ?" for k, op, _ in
[(c[0], c[1], c[2]) for c in self._conditions])
sql += f" WHERE {where}"
values = [c[2] for c in self._conditions]
if self._order_by:
sql += f" ORDER BY {self._order_by}"
if self._limit:
sql += f" LIMIT {self._limit}"
cursor = self._model._db.cursor()
cursor.execute(sql, values)
results = []
for row in cursor.fetchall():
instance = self._model(**dict(row))
results.append(instance)
return results
def get(self, **kwargs):
"""获取单个对象"""
results = self.filter(**kwargs).limit(1).all()
if not results:
raise ValueError("No matching object found")
return results[0]
def count(self) -> int:
"""返回匹配记录数"""
return len(self.all())
# ========== 使用示例 ==========
if __name__ == "__main__":
# 定义模型
class User(Model):
_table = "users"
id = IntegerField(primary_key=True)
name = TextField(nullable=False)
email = TextField(nullable=False)
age = IntegerField(default=0)
salary = RealField(default=0.0)
is_active = BooleanField(default=True)
class Order(Model):
_table = "orders"
id = IntegerField(primary_key=True)
user_id = IntegerField(nullable=False)
product = TextField(nullable=False)
amount = RealField(nullable=False)
# 连接数据库
Model.connect("test.db")
User.create_table()
Order.create_table()
# 创建记录
user = User(name="Alice", email="alice@example.com", age=30)
user.save()
user2 = User(name="Bob", email="bob@example.com", age=25, salary=50000)
user2.save()
# 查询
print("--- All Users ---")
for u in User.objects.all():
print(u, u.name, u.email)
print("\n--- Filter ---")
adults = User.objects.filter(age__gte=25).all()
for u in adults:
print(u, u.name, u.age)
print(f"\nTotal users: {User.objects.count()}")
# 更新
user.name = "Alice Updated"
user.save()
# 删除
user2.delete()
这个 ORM 引擎展示了元编程在实际框架设计中的核心应用：
- 元类 ModelMeta 在类定义时自动收集字段信息、设置表名、查找主键，避免了手动配置的冗余
- 字段描述符 Field 实现了属性访问时的值存储和脏标记追踪，使得 ORM 知道哪些字段需要保存
- 查询集 QuerySet 采用链式 API 设计，每次调用返回新的 QuerySet 实例，实现查询条件的组合
- __set_name__ 让描述符知道自己在类中的名称，无需手动传入
重点提示：
- 生产级 ORM（如 SQLAlchemy、Django ORM）还需要处理关系映射、事务管理、连接池等
- 描述符的 __get__ 在实例访问时返回值，在类访问时返回描述符本身
- SQL 参数化查询是防止 SQL 注入的关键，本实现始终使用占位符
- 元类 + 描述符 + __init_subclass__ 的组合可以构建非常灵活的框架
架构决策总结
Python 元编程提供了三个层次的动态性：__init_subclass__ 适用于简单的子类拦截；描述符协议适用于精细的属性控制；元类适用于完整的类创建控制。在实际的框架设计中，这三者往往组合使用：Django ORM 使用元类收集字段、使用描述符实现字段访问；SQLAlchemy 使用描述符实现 InstrumentedAttribute；FastAPI 使用描述符实现依赖注入。
元编程的核心理念是"用代码生成代码"。虽然过度使用元编程会降低可读性，但在框架层面，它是实现声明式 API、减少样板代码、提供统一抽象的关键技术。作为架构师，理解这些底层机制有助于更好地使用现有框架，以及在需要时构建定制化的基础设施。