web.py 實現類 Django ORM 查詢介面：Metaclass 動態注入 Manager、Model 基類與查詢封裝

Django ORM 的 Model.objects.all()、Model.objects.get() 等查詢介面極具表達力。本文示範在 web.py 框架中，透過 Metaclass 動態為模型類別注入 objects 管理器，實現類似 Django 的靜態查詢 API。

核心架構

# -*- coding: utf-8 -*-
import web
import config

def _connect_to_db():
    return web.database(dbn="sqlite", db=config.dbname)

class ModelError(Exception):
    def __init__(self, msg=""):
        self.msg = msg
    def __str__(self):
        return "ModelError: %s" % self.msg

# 查詢管理器
class ModelDefaultManager(object):
    def __init__(self, cls):
        self.cls = cls
        self._table_name = cls.__name__.lower()

    def all(self):
        db = _connect_to_db()
        results = db.select(self._table_name)
        return [self.cls(x) for x in results]

    def get(self, query_vars, where):
        results = self.filter(query_vars, where, limit=1)
        return results[0] if results else None

    def filter(self, query_vars, where, limit=None):
        db = _connect_to_db()
        try:
            results = db.select(self._table_name, vars=query_vars, where=where, limit=limit)
        except Exception as e:
            raise ModelError(str(e))
        return [self.cls(x) for x in results]

# Metaclass 動態注入 objects
class ModelMetaClass(type):
    def __new__(cls, classname, bases, attrs):
        new_class = super(ModelMetaClass, cls).__new__(cls, classname, bases, attrs)
        objects = ModelDefaultManager(new_class)
        setattr(new_class, "objects", objects)
        return new_class

# 基礎模型類別
class Model(object):
    __metaclass__ = ModelMetaClass

    def __init__(self):
        pass

    def _table_name(self):
        return self.__class__.__name__.lower()

    def insert(self, **kargs):
        db = _connect_to_db()
        try:
            with db.transaction():
                db.insert(self._table_name(), **kargs)
        except Exception as e:
            raise ModelError(str(e))

    def delete(self, where, using=None, vars=None):
        db = _connect_to_db()
        try:
            with db.transaction():
                db.delete(self._table_name(), where, vars=vars)
        except Exception as e:
            raise ModelError(str(e))

    def save(self, where, vars=None, **kargs):
        db = _connect_to_db()
        try:
            with db.transaction():
                db.update(self._table_name(), where, vars, **kargs)
        except Exception as e:
            raise ModelError(str(e))

使用方式

# 定義模型（無需實例化即可查詢）
class Users(Model):
    pass

# 類 Django 查詢 API
user_list = Users.objects.all()
user = Users.objects.get({"id": 1}, "id=$id")
users = Users.objects.filter({"name": "John"}, "name=$name", limit=10)

# 實例操作
u = Users()
u.insert(name="Alice", email="alice@example.com")
u.save("id=$id", {"id": 1}, name="Alice Updated")
u.delete("id=$id", vars={"id": 1})

關鍵設計點

Metaclass 工作流程

1. 定義 class Users(Model)
2. Python 呼叫 ModelMetaClass.__new__
3. 建立 Users 類別物件
4. 實例化 ModelDefaultManager(Users)
5. setattr(Users, "objects", manager)
6. 類別擁有 Users.objects 可直接查詢

擴充建議

• 支援關聯查詢 (ForeignKey, ManyToMany)
• 加入欄位定義與驗證 (CharField, IntegerField)
• 實現遷移機制
• 支援多資料庫路由
• 整合 SQLAlchemy 替代底層 web.database