这篇教程用 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='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:02class 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 -> Userbases -> ()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 -> Userbases -> ()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:02class 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 pymysqlclass 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 pymysqlclass 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知识宝库，欢迎大家来访。 以上就是用 Python 元类的特性实现 ORM 框架的详细内容，更多关于Python 实现 ORM 框架的资料请关注51zixue.net其它相关文章！
浅谈Python 中的复数问题
python中lstrip()截掉字符的实例讲解 |
