Python元类实现ORM框架

偏执的太偏执、 2022-10-20 12:16 186阅读 0赞

## 前言 ##

> 利用 `Python` 元类的特性实现 `ORM` 框架的 `insert` 功能，通过操作类对象，对数据表进行数据增加操作。
> 
> 由于 `ORM` 比较复杂，也不要重复造轮子，就完成一个 `insert` 相类似的ORM，理解其中的道理即可。

## ORM是什么 ##

O是 `object`，也就 **类对象** 的意思，R是 `relation`，翻译成中文是 **关系**，也就是关系数据库中 数据表 的意思，M是 `mapping`，是映射的意思。**在ORM框架中，它帮我们把类和数据表进行了一个映射，可以让我们通过类和类对象就能操作它所对应的表格中的数据**。ORM框架还有一个功能，它可以根据我们设计的类自动帮我们生成数据库中的表，省去了我们自己建表的过程。

一个句话理解就是：创建一个实例对象，用创建它的类名当做数据表名，用创建它的类属性对应数据表的字段，当对这个实例对象操作时，能够对应 `MySQL` 语句。

在 `Django` 中就内嵌了一个 `ORM` 框架，不需要直接面向数据库编程，而是定义模型类，通过模型类和对象完成数据表的增删改查操作。还有第三方库 `sqlalchemy` 都是 `ORM`框架。

![ORM映射关系][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()

## 测试功能 ##

### 准备数据库 ###

先准备数据库 `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** 表结构如下

### 创建模型类测试 ###

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

**✍ 码字不易，万水千山总是情，点赞再走行不行，还望各位大侠多多支持❤️**

## 公众号 ##

**新建文件夹X**

> 大自然用数百亿年创造出我们现实世界，而程序员用几百年创造出一个完全不同的虚拟世界。我们用键盘敲出一砖一瓦，用大脑构建一切。人们把1000视为权威，我们反其道行之，捍卫1024的地位。我们不是键盘侠，我们只是平凡世界中不凡的缔造者 。

[ORM]: /images/20221020/bdbf8041049c40d5aca967168e931e2c.png
[PythonKnowledge_ Python]: https://gitee.com/huiDBK/PythonKnowledge