新闻动态
新闻动态
NEWS INFORMATION

公司动态

产品资讯

行业资讯
轻云服务器升级CN2线路有什么优点
当前位置:   首页/开发运维 /用 Python 元类的特性实现 ORM 框架

用 Python 元类的特性实现 ORM 框架

发布日期：2022-06-09 14:13 | 文章来源：站长之家

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知识宝库，欢迎大家来访。

以上就是用 Python 元类的特性实现 ORM 框架的详细内容，更多关于Python 实现 ORM 框架的资料请关注本站其它相关文章！

香港稳定服务器

版权声明：本站文章来源标注为YINGSOO的内容版权均为本站所有，欢迎引用、转载，请保持原文完整并注明来源及原文链接。禁止复制或仿造本网站，禁止在非www.yingsoo.com所属的服务器上建立镜像，否则将依法追究法律责任。本站部分内容来源于网友推荐、互联网收集整理而来，仅供学习参考，不代表本站立场，如有内容涉嫌侵权，请联系alex-e#qq.com处理。

相关文章

实时开通

自选配置、实时开通

免备案

全球线路精选！

全天候客户服务

7x24全年不间断在线

专属顾问服务

1对1客户咨询顾问

在线
客服

在线客服：7*24小时在线

客服
热线

400-630-3752
7*24小时客服服务热线

关注
微信

关注官方微信
顶部