Python 元類程式設計實現一個簡單的 ORM
阿新 • • 發佈：2020-11-19
概述
什麼是ORM?
ORM全稱“Object Relational Mapping”，即物件-關係對映，就是把關係資料庫的一行對映為一個物件，也就是一個類對應一個表，這樣，寫程式碼更簡單，不用直接操作SQL語句。
現在我們就要實現簡易版ORM。
效果
class Person(Model): """ 定義類的屬性到列的對映 """ pid = IntegerField('id') names = StringField('username') email = StringField('email') password= StringField('password') p = Person(pid=10086, names='曉明', email='[email protected]', password='123456') p.save()
通過執行save()方法 動態生成sql插入語句, 是不是很神奇， 那我們現在開始解析原理吧
步驟
首先我們要定義一個 Field 類 它負責儲存資料庫表的欄位名和欄位型別：
class Field(object): def __init__(self, name, column_type): self.name = name self.column_type= column_type def __str__(self): return '<%s:%s>' % (self.__class__.__name__, self.name)
在Field的基礎上，進一步定義各種型別的Field，比如StringField，IntegerField等等：
class StringField(Field): def __init__(self, name): super(StringField, self).__init__(name, 'varchar(100)') class IntegerField(Field):def __init__(self, name): super(IntegerField, self).__init__(name, 'bigint')
下一步，就是編寫最複雜的ModelMetaclass：
class ModelMetaclass(type): def __new__(cls, name, bases, attrs): if name == "Model": return type.__new__(cls, name, bases, attrs) mappings = dict() print("Found class: %s" % name) for k, v in attrs.items(): if isinstance(v, Field): print("Found mapping: %s ==> %s" % (k, v)) mappings[k] = v for k in mappings.keys(): attrs.pop(k) attrs["__table__"] = name # 表名和類名一致 attrs["__mappings__"] = mappings # 儲存屬性和列的對映關係 return type.__new__(cls, name, bases, attrs)
最後就是基類Model:
class Model(metaclass=ModelMetaclass): def __init__(self, **kwargs): _setattr = setattr if kwargs: for k, v in kwargs.items(): _setattr(self, k, v) super(Model, self).__init__() def save(self): fields = [] params = [] args = [] for k, v in self.__mappings__.items(): fields.append(k) params.append("?") args.append(getattr(self, k, None)) sql = "insert into %s (%s) values (%s)" % (self.__table__, ','.join(fields), ",".join(params)) print('插入語句: %s' % sql) print('引數: %s' % str(args)) def update(self): fields = [] args = [] for k, v in self.__mappings__.items(): if getattr(self, k, None): fields.append(k+"=?") args.append(getattr(self, k, None)) sql = "update %s set %s" % (self.__table__, ','.join(fields)) print("更新語句: %s " % sql) print("引數: %s" % args) def filter(self, *args): pass def delete(self): pass
當用戶定義一個 class Person(Model) 繼承父類時，Python直譯器會在當前類Person的定義中找__metaclass__，如果沒有找到，就繼續到父類中找__metaclass__，實在找不到就用預設 type 類。
我們在父類Model中定義了__metaclass__的 ModelMetaclass來建立 Person 類，所以 metaclass 隱式地繼承到子類。
在ModelMetaclass中，一共做了幾件事情：
-
排除掉對
Model類的修改； -
在當前類（比如
Person）中查詢定義的類的所有屬性，如果找到一個 Field 屬性，就把它儲存到一個__mappings__的dict中，同時從類屬性中刪除該Field屬性，否則，容易造成執行時錯誤； -
把表名儲存到
__table__中，這裡簡化為表名預設為類名。
在Model類中，就可以定義各種操作資料庫的方法，比如save()，delete()，find()，update()等等。
我們實現了save(), update()方法，把一個例項儲存到資料庫中。因為有表名，屬性到欄位的對映和屬性值的集合，就可以構造出INSERT語句和UPDATE語句。
編寫程式碼試試：
class UserInfo(Model): """ 定義類的屬性到列的對映 """ uid = IntegerField('uid') name = StringField('username') email = StringField('email') password = StringField('password') class Person(Model): """ 定義類的屬性到列的對映 """ pid = IntegerField('id') names = StringField('username') email = StringField('email') password = StringField('password') p = Person(pid=10086, names='曉明', email='[email protected]', password='123456') p.save() u2 = UserInfo(password='123456') u2.update()
輸出
Found class: UserInfo Found mapping: uid ==> <IntegerField:uid> Found mapping: name ==> <StringField:username> Found mapping: email ==> <StringField:email> Found mapping: password ==> <StringField:password> Found class: Person Found mapping: pid ==> <IntegerField:id> Found mapping: names ==> <StringField:username> Found mapping: email ==> <StringField:email> Found mapping: password ==> <StringField:password> 插入語句: insert into Person (pid,names,email,password) values (?,?,?,?) 引數: [10086, '曉明', '[email protected]', '123456'] 更新語句: update UserInfo set password=? 引數: ['123456']
結束語
就這樣一個小巧的ORM就這麼完成了。是不是學到了很多呢 ？這裡利用的是超程式設計，很多Python框架都運用了超程式設計達到動態操作類。
注：上述程式碼列子 結合了廖雪峰的列子和少量的django ORM原始碼。
完整程式碼
class Field(object): def __init__(self, name, column_type): self.name = name self.column_type = column_type def __str__(self): return '<%s:%s>' % (self.__class__.__name__, self.name) class StringField(Field): def __init__(self, name): super(StringField, self).__init__(name, 'varchar(100)') class IntegerField(Field): def __init__(self, name): super(IntegerField, self).__init__(name, 'bigint') class ModelMetaclass(type): def __new__(cls, name, bases, attrs): if name == "Model": return type.__new__(cls, name, bases, attrs) mappings = dict() print("Found class: %s" % name) for k, v in attrs.items(): if isinstance(v, Field): print("Found mapping: %s ==> %s" % (k, v)) mappings[k] = v for k in mappings.keys(): attrs.pop(k) attrs["__table__"] = name # 表名和類名一致 attrs["__mappings__"] = mappings # 儲存屬性和列的對映關係 return type.__new__(cls, name, bases, attrs) class Model(metaclass=ModelMetaclass): def __init__(self, **kwargs): _setattr = setattr if kwargs: for k, v in kwargs.items(): _setattr(self, k, v) super(Model, self).__init__() def save(self): fields = [] params = [] args = [] for k, v in self.__mappings__.items(): fields.append(k) params.append("?") args.append(getattr(self, k, None)) sql = "insert into %s (%s) values (%s)" % (self.__table__, ','.join(fields), ",".join(params)) print('插入語句: %s' % sql) print('引數: %s' % str(args)) def update(self): fields = [] args = [] for k, v in self.__mappings__.items(): if getattr(self, k, None): fields.append(k+"=?") args.append(getattr(self, k, None)) sql = "update %s set %s" % (self.__table__, ','.join(fields)) print("更新語句: %s " % sql) print("引數: %s" % args) def filter(self, *args): pass def delete(self): pass class UserInfo(Model): """ 定義類的屬性到列的對映 """ uid = IntegerField('uid') name = StringField('username') email = StringField('email') password = StringField('password') class Person(Model): """ 定義類的屬性到列的對映 """ pid = IntegerField('id') names = StringField('username') email = StringField('email') password = StringField('password') p = Person(pid=10086, names='曉明', email='[email protected]', password='123456') p.save() u2 = UserInfo(password='123456') u2.update()