使用Python的web.py框架實現類似Django的ORM查詢的教程

來源：懂視網責編：小采時間：2020-11-27 14:41:46

使用Python的web.py框架實現類似Django的ORM查詢的教程

使用Python的web.py框架實現類似Django的ORM查詢的教程:Django中的對象查詢 Django框架自帶了ORM，實現了一些比較強大而且方便的查詢功能，這些功能和表無關。比如下面這個例子： class Question(models.Model): question_text = models.CharField(max_length=200) pub_date

推薦度：

點擊下載本文 文檔為doc格式

導讀使用Python的web.py框架實現類似Django的ORM查詢的教程:Django中的對象查詢 Django框架自帶了ORM，實現了一些比較強大而且方便的查詢功能，這些功能和表無關。比如下面這個例子： class Question(models.Model): question_text = models.CharField(max_length=200) pub_date

Django中的對象查詢

Django框架自帶了ORM，實現了一些比較強大而且方便的查詢功能，這些功能和表無關。比如下面這個例子：

class Question(models.Model):
 question_text = models.CharField(max_length=200)
 pub_date = models.DateTimeField('date published')


>>> Question.objects.all()
>>> Question.objects.get(pk=1)

從例子可以看出，objects.all和objects.get這些功能都不是在class Question中定義的，可能在其父類models.Model中定義，也可能不是。那么我們在web.py中如何實現這樣的功能呢？（如果你選擇使用SQLAlchemy就不需要自己實現了）。
實現
思路

我們注意到Question.objects.all()這樣的調用是直接訪問了類屬性objects，并調用了objects屬性的方法all()。這里objects可能是一個實例，也可能是一個類。我個人認為（我沒看過Django的實現）這應該是一個實例，因為實例化的過程可以傳遞一些表的信息，使得類似all()這樣的函數可以工作。經過分析之后，我們可以列出我們需要解決的問題：

需要實現一個模型的父類Model，實際的表可以從這個父類繼承以獲得自己沒有定義的功能。

實際的模型類（比如Question類）定義后，不實例話的情況下就要具備objects.all()這樣的查詢效果。

從上面的需求可以看出，我們需要在類定義的時候就實現這些功能，而不是等到類實例化的時候再實現這些功能。類定義的時候實現功能？這不就是metaclass（元類）做的事情嘛。因此實現過程大概是下面這樣的：

實現一個Model類，其綁定方法和表的增、刪、改有關。

修改Model類的元類為ModelMetaClass，該元類定義的過程中為類增加一個objects對象，該對象是一個ModelDefaultManager類的實例，實現了表的查詢功能。

代碼

都說不給代碼就是耍流氓，我還是給吧。說明下：使用的數據庫操作都是web.py的db庫中的接口。

 # -*- coding: utf-8 -*-

 import web

 import config # 自定義的配置類，可以忽略


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


 def init_db():
 db = _connect_to_db()
 for statement in config.sql_statements:
 db.query(statement)


 class ModelError(Exception):
 """Exception raised by all models.

 Attributes:
 msg: Error message.
 """

 def __init__(self, msg=""):
 self.msg = msg

 def __str__(self):
 return "ModelError: %s" % self.msg


 class ModelDefaultManager(object):
 """ModelManager implements query functions against a model.

 Attributes:
 cls: The class to be managed.
 """

 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)
 if len(results) > 0:
 return results[0]
 else:
 return 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]


 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):
 """Parent class of all models.
 """

 __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))

使用

首先定義表對應的類：