用 Python 元类的特性实现 ORM 框架
作者：忆想不到的晖 发布时间：2022-02-12 12:45:24
标签：python,元类,orm框架
目录
ORM是什么
实现ORM中的insert功能
完善对数据类型的检测
抽取到基类中
添加数据库驱动执行sql语句
添加数据库驱动执行sql语句
测试功能
准备数据库
创建模型类测试
源代码
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='huidbk@163.com', password='123456')
user.save()
# 对应如下sql语句
# insert into User (uid,username,email,password) values (123,hui,huidbk@163.com,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='huidbk@163.com', 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,huidbk@163.com,123456)
完善对数据类型的检测
上面转成的 sql 语句如下：
insert into User (uid,username,email,password) values (12345,hui,huidbk@163.com,123456)
发现没有，在 sql 语句中字符串类型没有没有引号 ''
正确的 sql 语句应该是：
insert into User (uid,username,email,password) values (123, 'hui', 'huidbk@163.com', '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='huidbk@163.com', 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','huidbk@163.com','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='huidbk@163.com', 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='huidbk@163.com', 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 | huidbk@163.com | 123456 |
| 2 | name2 | huidbk@162.com | 123452 |
| 3 | name3 | huidbk@163.com | 123453 |
| 4 | name4 | huidbk@164.com | 123454 |
| 5 | name5 | huidbk@165.com | 123455 |
| 6 | name6 | huidbk@166.com | 123456 |
| 7 | name7 | huidbk@167.com | 123457 |
| 8 | name8 | huidbk@168.com | 123458 |
| 9 | name9 | huidbk@169.com | 123459 |
+-----+----------+----------------+----------+
9 rows in set (0.00 sec)
源代码
源代码已上传到 Gitee PythonKnowledge: Python知识宝库，欢迎大家来访。
来源：https://juejin.cn/post/6963443372266618917
0
投稿猜你喜欢
在做数据分析或者统计的时候，经常需要进行数据正态性的检验，因为很多假设都是基于正态分布的基础之上的，例如：T检验。在Python中，主要有以- 研究了一段时间酷狗音乐的接口，完美破解了其vip音乐下载方式，想着能更好的追求开源，故写下此篇文章，本文仅供学习参考。虽然没什么
- 本文实例讲述了Python批量重命名同一文件夹下文件的方法。分享给大家供大家参考。具体分析如下：朋友发了一个文件夹过来，里面的图片都以 .t
- 一、时间获取函数>>> import time>>> time.time()1570530861.740
- 获取所有variable（每个op中可训练的张量）的name：for variable_name in tf.global_variable
- 起因是因为公司要开发一款自动登录某网站的助手工具提供给客户使用，要使用到selenium，所以选择了pyqt5的方式来开发这个C/S架构的客
- Python应用编程需要用到的针对不同数据库引擎的数据库接口：http://wiki.python.org/moin/DatabaseInt
- 这两天看了下某位大神的github,知道他对算法比较感兴趣,看了其中的一个计算数字的步数算法,感觉这个有点意思,所以就自己实现了一个。算法描
- 场景go 如果频繁地创建、销毁对象（比如 http 服务的 json 对象，日志内容等），会对 GC 造成压力。比如下面的 Log 函数，在
- 本文实例讲述了Python实现给文件添加内容及得到文件信息的方法。分享给大家供大家参考。具体分析如下：经常会遇到给文件添加内容的时候，如果只
- 本文讲解如何设置SQL Server数据库全文索引服务。在Microsoft SQL Server 7.0 中提供了全文索引服务(Full-
- 参考官网地址：Windows端：https://tensorflow.google.cn/install/source_windowsCPU
- URL重定向行为路由的尾部加不加斜杠不一样的，比如：from flask import Flaskapp = Flask(__name__)
- 调用函数：#!/usr/bin/env python3 # -*- coding: utf-8 -*- # 函数调用 >>>
- 一.应用场景多个组件共用一个方法时可以用 mixin 抽取到一个js文件中，作为共用方法二.实现方法1.提取js共用方法文件export c
- 目录背景认识复合索引最左匹配原则字段顺序的影响复合索引可以替代单一索引吗？小结背景最近频繁出现慢SQL导致系统性能问题，于是决定针对索引进行
- 求和try: while True: n=input() s=1 for x in raw_input(
- 1.概述随着人工智能技术的不断发展，越来越多的AI产品被应用到各个领域，其中最具代表性的莫过于人工智能语言模型。语言模型是一种可以通过学习大
- 文件可以传输，但是对比传输前后的文件：socket_test.txt，末尾有一些不一致服务端代码：#!/usr/bin/python# -*
- 代码如下import wordcloudimport jiebafont = r'C:\Windows\Fonts\simfang.
