用 Python 元類的特性實現 ORM 框架
ORM是什麼
O是 object，也就 類對象 的意思，R是 relation，翻譯成中文是 關系，也就是關系數據庫中 數據表 的意思，M是 mapping，是映射的意思。在ORM框架中，它幫我們把類和數據表進行瞭一個映射，可以讓我們通過類和類對象就能操作它所對應的表格中的數據。ORM框架還有一個功能，它可以根據我們設計的類自動幫我們生成數據庫中的表，省去瞭我們自己建表的過程。
一個句話理解就是：創建一個實例對象，用創建它的類名當做數據表名，用創建它的類屬性對應數據表的字段，當對這個實例對象操作時，能夠對應 MySQL 語句。
在 Django 中就內嵌瞭一個 ORM 框架，不需要直接面向數據庫編程，而是定義模型類，通過模型類和對象完成數據表的增刪改查操作。還有第三方庫 sqlalchemy 都是 ORM框架。
先看看我們大致要實現什麼功能
class User(父類省略):
uid = ('uid', "int unsigned")
name = ('username', "varchar(30)")
email = ('email', "varchar(30)")
password = ('password', "varchar(30)")
...省略...
user = User(uid=123, name='hui', email='[email protected]', password='123456')
user.save()
# 對應如下sql語句
# insert into User (uid,username,email,password) values (123,hui,[email protected],123456)
所謂的 ORM 就是讓開發者在操作數據庫的時候，能夠像操作對象時通過xxxx.屬性=yyyy一樣簡單，這是開發ORM的初衷。
實現ORM中的insert功能
通過 Python 中 元類 簡單實現 ORM 中的 insert 功能
# !/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 利用Python元類簡單實現ORM框架的Insert插入功能 }
# @Date: 2021/05/17 17:02
class ModelMetaclass(type):
"""數據表模型元類"""
def __new__(mcs, cls_name, bases, attrs):
print(f'cls_name -> {cls_name}') # 類名
print(f'bases -> {bases}') # 繼承類
print(f'attrs -> {attrs}') # 類中所有屬性
print()
# 數據表對應關系字典
mappings = dict()
# 過濾出對應數據表的字段屬性
for k, v in attrs.items():
# 判斷是否是指定的StringField或者IntegerField的實例對象
# 這裡就簡單判斷字段是元組
if isinstance(v, tuple):
print('Found mapping: %s ==> %s' % (k, v))
mappings[k] = v
# 刪除這些已經在字典中存儲的字段屬性
for k in mappings.keys():
attrs.pop(k)
# 將之前的uid/name/email/password以及對應的對象引用、類名字
# 用其他類屬性名稱保存
attrs['__mappings__'] = mappings # 保存屬性和列的映射關系
attrs['__table__'] = cls_name # 假設表名和類名一致
return type.__new__(mcs, cls_name, bases, attrs)
class User(metaclass=ModelMetaclass):
"""用戶模型類"""
# 類屬性名 表字段 表字段類型
uid = ('uid', 'int unsigned')
name = ('username', 'varchar(30)')
email = ('email', 'varchar(30)')
password = ('password', 'varchar(30)')
def __init__(self, **kwargs):
for name, value in kwargs.items():
setattr(self, name, value)
def save(self):
fields = []
args = []
for k, v in self.__mappings__.items():
fields.append(v[0])
args.append(getattr(self, k, None))
# 表名
table_name = self.__table__
# 數據表中的字段
fields = ','.join(fields)
# 待插入的數據
args = ','.join([str(i) for i in args])
# 生成sql語句
sql = f"""insert into {table_name} ({fields}) values ({args})"""
print(f'SQL: {sql}')
def main():
user = User(uid=123, name='hui', email='[email protected]', password='123456')
user.save()
if __name__ == '__main__':
main()
當 User 指定元類之後，uid、name、email、password 類屬性將不在類中，而是在 __mappings__ 屬性指定的字典中存儲。 User 類的這些屬性將轉變為如下
__mappings__ = {
"uid": ('uid', "int unsigned")
"name": ('username', "varchar(30)")
"email": ('email', "varchar(30)")
"password": ('password', "varchar(30)")
}
__table__ = "User"
執行的效果如下：
cls_name -> User
bases -> ()
attrs -> {
'__module__': '__main__', '__qualname__': 'User', '__doc__': '用戶模型類',
'uid': ('uid', 'int unsigned'),
'name': ('username', 'varchar(30)'),
'email': ('email', 'varchar(30)'),
'password': ('password', 'varchar(30)'),
'__init__': <function User.__init__ at 0x0000026D520C1048>,
'save': <function User.save at 0x0000026D520C10D8>
}
Found mapping: uid ==> ('uid', 'int unsigned')
Found mapping: name ==> ('username', 'varchar(30)')
Found mapping: email ==> ('email', 'varchar(30)')
Found mapping: password ==> ('password', 'varchar(30)')
SQL: insert into User (uid,username,email,password) values (123,hui,[email protected],123456)
完善對數據類型的檢測
上面轉成的 sql 語句如下：
insert into User (uid,username,email,password) values (12345,hui,[email protected],123456)
發現沒有，在 sql 語句中字符串類型沒有沒有引號 ”
正確的 sql 語句應該是：
insert into User (uid,username,email,password) values (123, 'hui', '[email protected]', '123456')
因此修改 User 類完善數據類型的檢測
class ModelMetaclass(type):
# 此處和上文一樣, 故省略....
pass
class User(metaclass=ModelMetaclass):
"""用戶模型類"""
uid = ('uid', "int unsigned")
name = ('username', "varchar(30)")
email = ('email', "varchar(30)")
password = ('password', "varchar(30)")
def __init__(self, **kwargs):
for name, value in kwargs.items():
setattr(self, name, value)
# 在這裡完善數據類型檢測
def save(self):
fields = []
args = []
for k, v in self.__mappings__.items():
fields.append(v[0])
args.append(getattr(self, k, None))
# 把參數數據類型對應數據表的字段類型
args_temp = list()
for temp in args:
if isinstance(temp, int):
args_temp.append(str(temp))
elif isinstance(temp, str):
args_temp.append(f"'{temp}'")
# 表名
table_name = self.__table__
# 數據表中的字段
fields = ','.join(fields)
# 待插入的數據
args = ','.join(args_temp)
# 生成sql語句
sql = f"""insert into {table_name} ({fields}) values ({args})"""
print(f'SQL: {sql}')
def main():
user = User(uid=123, name='hui', email='[email protected]', password='123456')
user.save()
if __name__ == '__main__':
main()
運行效果如下：
cls_name -> User
bases -> ()
attrs -> {
'__module__': '__main__', '__qualname__': 'User', '__doc__': '用戶模型類',
'uid': ('uid', 'int unsigned'),
'name': ('username', 'varchar(30)'),
'email': ('email', 'varchar(30)'),
'password': ('password', 'varchar(30)'),
'__init__': <function User.__init__ at 0x0000026D520C1048>,
'save': <function User.save at 0x0000026D520C10D8>
}
Found mapping: uid ==> ('uid', 'int unsigned')
Found mapping: name ==> ('username', 'varchar(30)')
Found mapping: email ==> ('email', 'varchar(30)')
Found mapping: password ==> ('password', 'varchar(30)')
SQL: insert into User (uid,username,email,password) values(123,'hui','[email protected]','123456')
抽取到基類中
# !/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 利用Python元類實現ORM框架的Insert插入功能 }
# @Date: 2021/05/17 17:02
class ModelMetaclass(type):
"""數據表模型元類"""
def __new__(mcs, cls_name, bases, attrs):
print(f'cls_name -> {cls_name}') # 類名
print(f'bases -> {bases}') # 繼承類
print(f'attrs -> {attrs}') # 類中所有屬性
print()
# 數據表對應關系字典
mappings = dict()
# 過濾出對應數據表的字段屬性
for k, v in attrs.items():
# 判斷是否是對應數據表的字段屬性, 因為attrs中包含所有的類屬性
# 這裡就簡單判斷字段是元組
if isinstance(v, tuple):
print('Found mapping: %s ==> %s' % (k, v))
mappings[k] = v
# 刪除這些已經在字典中存儲的字段屬性
for k in mappings.keys():
attrs.pop(k)
# 將之前的uid/name/email/password以及對應的對象引用、類名字
# 用其他類屬性名稱保存
attrs['__mappings__'] = mappings # 保存屬性和列的映射關系
attrs['__table__'] = cls_name # 假設表名和類名一致
return type.__new__(mcs, cls_name, bases, attrs)
class Model(object, metaclass=ModelMetaclass):
"""數據表模型基類"""
def __init__(self, **kwargs):
for name, value in kwargs.items():
setattr(self, name, value)
def save(self):
fields = []
args = []
for k, v in self.__mappings__.items():
fields.append(v[0])
args.append(getattr(self, k, None))
# 把參數數據類型對應數據表的字段類型
args_temp = list()
for temp in args:
if isinstance(temp, int):
args_temp.append(str(temp))
elif isinstance(temp, str):
args_temp.append(f"'{temp}'")
# 表名
table_name = self.__table__
# 數據表中的字段
fields = ','.join(fields)
# 待插入的數據
args = ','.join(args_temp)
# 生成sql語句
sql = f"""insert into {table_name} ({fields}) values ({args})"""
print(f'SQL: {sql}')
# 執行sql語句
# ...
class User(Model):
"""用戶表模型類"""
uid = ('uid', "int unsigned")
name = ('username', "varchar(30)")
email = ('email', "varchar(30)")
password = ('password', "varchar(30)")
def main():
user = User(uid=123, name='hui', email='[email protected]', password='123456')
user.save()
if __name__ == '__main__':
main()
添加數據庫驅動執行sql語句
這裡我們使用 pymysql 數據庫驅動，來執行 sql 語句
在 Model 類中新增一個 get_connection 的靜態方法用於獲取數據庫連接
import pymysql
class Model(object, metaclass=ModelMetaclass):
"""數據表模型基類"""
def __init__(self, **kwargs):
for name, value in kwargs.items():
setattr(self, name, value)
@staticmethod
def get_connection():
"""
獲取數據庫連接與數據遊標
:return: conn, cursor
"""
conn = pymysql.connect(
database='testdb',
host='localhost',
port=3306,
user='root',
password='123456'
)
return conn, conn.cursor()
def save(self):
fields = []
args = []
for k, v in self.__mappings__.items():
fields.append(v[0])
args.append(getattr(self, k, None))
# 把參數數據類型對應數據表的字段類型
args_temp = list()
for temp in args:
if isinstance(temp, int):
args_temp.append(str(temp))
elif isinstance(temp, str):
args_temp.append(f"'{temp}'")
# 表名
table_name = self.__table__
# 數據表中的字段
fields = ','.join(fields)
# 待插入的數據
args = ','.join(args_temp)
# 生成sql語句
sql = f"""insert into {table_name} ({fields}) values ({args})"""
print(f'SQL: {sql}')
# 執行sql語句
conn, cursor = self.get_connection()
ret = cursor.execute(sql)
print(ret)
conn.commit()
cursor.close()
conn.close()
添加數據庫驅動執行sql語句
這裡我們使用 pymysql 數據庫驅動，來執行 sql 語句
在 Model 類中新增一個 get_connection 的靜態方法用於獲取數據庫連接
import pymysql
class Model(object, metaclass=ModelMetaclass):
"""數據表模型基類"""
def __init__(self, **kwargs):
for name, value in kwargs.items():
setattr(self, name, value)
@staticmethod
def get_connection():
"""
獲取數據庫連接與數據遊標
:return: conn, cursor
"""
conn = pymysql.connect(
database='testdb',
host='localhost',
port=3306,
user='root',
password='123456'
)
return conn, conn.cursor()
def save(self):
fields = []
args = []
for k, v in self.__mappings__.items():
fields.append(v[0])
args.append(getattr(self, k, None))
# 把參數數據類型對應數據表的字段類型
args_temp = list()
for temp in args:
if isinstance(temp, int):
args_temp.append(str(temp))
elif isinstance(temp, str):
args_temp.append(f"'{temp}'")
# 表名
table_name = self.__table__
# 數據表中的字段
fields = ','.join(fields)
# 待插入的數據
args = ','.join(args_temp)
# 生成sql語句
sql = f"""insert into {table_name} ({fields}) values ({args})"""
print(f'SQL: {sql}')
# 執行sql語句
conn, cursor = self.get_connection()
ret = cursor.execute(sql)
print(ret)
conn.commit()
cursor.close()
conn.close()
測試功能
準備數據庫
先準備數據庫 testdb 和 user 數據表
create database testdb charset=utf8; use testdb; create table user( uid int unsigned auto_increment primary key, username varchar(30) not null, email varchar(30), password varchar(30) not null );
user 表結構如下
+----------+------------------+------+-----+---------+----------------+ | Field | Type | Null | Key | Default | Extra | +----------+------------------+------+-----+---------+----------------+ | uid | int(10) unsigned | NO | PRI | NULL | auto_increment | | username | varchar(30) | NO | | NULL | | | email | varchar(30) | YES | | NULL | | | password | varchar(30) | NO | | NULL | | +----------+------------------+------+-----+---------+----------------+
創建模型類測試
class User(Model):
"""用戶表模型類"""
uid = ('uid', "int unsigned")
name = ('username', "varchar(30)")
email = ('email', "varchar(30)")
password = ('password', "varchar(30)")
def main():
user = User(uid=1, name='hui', email='[email protected]', password='123456')
user.save()
for i in range(2, 10):
user = User(
uid=i,
name=f'name{i}',
email=f'huidbk@16{i}.com',
password=f'12345{i}'
)
user.save()
if __name__ == '__main__':
main()
查看數據庫 user 表數據
mysql> select * from user; +-----+----------+----------------+----------+ | uid | username | email | password | +-----+----------+----------------+----------+ | 1 | hui | [email protected] | 123456 | | 2 | name2 | [email protected] | 123452 | | 3 | name3 | [email protected] | 123453 | | 4 | name4 | [email protected] | 123454 | | 5 | name5 | [email protected] | 123455 | | 6 | name6 | [email protected] | 123456 | | 7 | name7 | [email protected] | 123457 | | 8 | name8 | [email protected] | 123458 | | 9 | name9 | [email protected] | 123459 | +-----+----------+----------------+----------+ 9 rows in set (0.00 sec)
源代碼
源代碼已上傳到 Gitee PythonKnowledge: Python知識寶庫，歡迎大傢來訪。
以上就是用 Python 元類的特性實現 ORM 框架的詳細內容，更多關於Python 實現 ORM 框架的資料請關註WalkonNet其它相關文章！
推薦閱讀：
- Oracle多表查詢中間表的創建實例教程
- Python操作PostgreSql數據庫的方法(基本的增刪改查)
- MySQL約束(創建表時的各種條件說明)
- MySQL數據庫如何給表設置約束詳解
- 基於Python實現一個簡易的數據管理系統