App下載

Python ORM 框架怎么實(shí)現(xiàn)？

猿友 2021-07-27 09:46:59 瀏覽數(shù) (2398)
反饋

ORM是什么

O是 object，也就 類對(duì)象 的意思，R是 relation，翻譯成中文是 關(guān)系，也就是關(guān)系數(shù)據(jù)庫(kù)中 數(shù)據(jù)表 的意思，M是 mapping，是映射的意思。在ORM框架中，它幫我們把類和數(shù)據(jù)表進(jìn)行了一個(gè)映射，可以讓我們通過(guò)類和類對(duì)象就能操作它所對(duì)應(yīng)的表格中的數(shù)據(jù)。ORM框架還有一個(gè)功能，它可以根據(jù)我們?cè)O(shè)計(jì)的類自動(dòng)幫我們生成數(shù)據(jù)庫(kù)中的表，省去了我們自己建表的過(guò)程。

一個(gè)句話理解就是：創(chuàng)建一個(gè)實(shí)例對(duì)象，用創(chuàng)建它的類名當(dāng)做數(shù)據(jù)表名，用創(chuàng)建它的類屬性對(duì)應(yīng)數(shù)據(jù)表的字段，當(dāng)對(duì)這個(gè)實(shí)例對(duì)象操作時(shí)，能夠?qū)?yīng) MySQL 語(yǔ)句。

在 Django 中就內(nèi)嵌了一個(gè) ORM 框架，不需要直接面向數(shù)據(jù)庫(kù)編程，而是定義模型類，通過(guò)模型類和對(duì)象完成數(shù)據(jù)表的增刪改查操作。還有第三方庫(kù) sqlalchemy 都是 ORM框架。

對(duì)象與數(shù)據(jù)表的一一映射

先看看我們大致要實(shí)現(xiàn)什么功能

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()

# 對(duì)應(yīng)如下sql語(yǔ)句
# insert into User (uid,username,email,password) values (123,hui,huidbk@163.com,123456)

所謂的 ORM 就是讓開發(fā)者在操作數(shù)據(jù)庫(kù)的時(shí)候，能夠像操作對(duì)象時(shí)通過(guò)xxxx.屬性=yyyy一樣簡(jiǎn)單，這是開發(fā)ORM的初衷。

實(shí)現(xiàn)ORM中的insert功能

通過(guò) Python 中 元類 簡(jiǎn)單實(shí)現(xiàn) ORM 中的 insert 功能

# !/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 利用Python元類簡(jiǎn)單實(shí)現(xiàn)ORM框架的Insert插入功能 }
# @Date: 2021/05/17 17:02


class ModelMetaclass(type):
    """數(shù)據(jù)表模型元類"""

    def __new__(mcs, cls_name, bases, attrs):

        print(f'cls_name -> {cls_name}')    # 類名
        print(f'bases -> {bases}')          # 繼承類
        print(f'attrs -> {attrs}')          # 類中所有屬性
        print()

        # 數(shù)據(jù)表對(duì)應(yīng)關(guān)系字典
        mappings = dict()

        # 過(guò)濾出對(duì)應(yīng)數(shù)據(jù)表的字段屬性
        for k, v in attrs.items():
            # 判斷是否是指定的StringField或者IntegerField的實(shí)例對(duì)象
            # 這里就簡(jiǎn)單判斷字段是元組
            if isinstance(v, tuple):
                print('Found mapping: %s ==> %s' % (k, v))
                mappings[k] = v

        # 刪除這些已經(jīng)在字典中存儲(chǔ)的字段屬性
        for k in mappings.keys():
            attrs.pop(k)

        # 將之前的uid/name/email/password以及對(duì)應(yīng)的對(duì)象引用、類名字
        # 用其他類屬性名稱保存
        attrs['__mappings__'] = mappings  # 保存屬性和列的映射關(guān)系
        attrs['__table__'] = cls_name     # 假設(shè)表名和類名一致
        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__
        # 數(shù)據(jù)表中的字段
        fields = ','.join(fields)
        # 待插入的數(shù)據(jù)
        args = ','.join([str(i) for i in args])
        
        # 生成sql語(yǔ)句
        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()

當(dāng) User 指定元類之后，uid、name、email、password 類屬性將不在類中，而是在 __mappings__ 屬性指定的字典中存儲(chǔ)。 User 類的這些屬性將轉(zhuǎn)變?yōu)槿缦?/p> 

__mappings__ = {
    "uid": ('uid', "int unsigned")
    "name": ('username', "varchar(30)")
    "email": ('email', "varchar(30)")
    "password": ('password', "varchar(30)")
}
__table__ = "User"

執(zhí)行的效果如下：

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)

完善對(duì)數(shù)據(jù)類型的檢測(cè)

上面轉(zhuǎn)成的 sql 語(yǔ)句如下：

insert into User (uid,username,email,password) values (12345,hui,huidbk@163.com,123456)

發(fā)現(xiàn)沒有，在 sql 語(yǔ)句中字符串類型沒有沒有引號(hào) ''

正確的 sql 語(yǔ)句應(yīng)該是：

insert into User (uid,username,email,password) values (123, 'hui', 'huidbk@163.com', '123456')

因此修改 User 類完善數(shù)據(jù)類型的檢測(cè)

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)

    # 在這里完善數(shù)據(jù)類型檢測(cè)
    def save(self):
        fields = []
        args = []
        for k, v in self.__mappings__.items():
            fields.append(v[0])
            args.append(getattr(self, k, None))

        # 把參數(shù)數(shù)據(jù)類型對(duì)應(yīng)數(shù)據(jù)表的字段類型
        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__
        # 數(shù)據(jù)表中的字段
        fields = ','.join(fields)
        # 待插入的數(shù)據(jù)
        args = ','.join(args_temp)

        # 生成sql語(yǔ)句
        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()

運(yùn)行效果如下：

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元類實(shí)現(xiàn)ORM框架的Insert插入功能 }
# @Date: 2021/05/17 17:02


class ModelMetaclass(type):
    """數(shù)據(jù)表模型元類"""

    def __new__(mcs, cls_name, bases, attrs):

        print(f'cls_name -> {cls_name}')  # 類名
        print(f'bases -> {bases}')  # 繼承類
        print(f'attrs -> {attrs}')  # 類中所有屬性
        print()

        # 數(shù)據(jù)表對(duì)應(yīng)關(guān)系字典
        mappings = dict()

        # 過(guò)濾出對(duì)應(yīng)數(shù)據(jù)表的字段屬性
        for k, v in attrs.items():
            # 判斷是否是對(duì)應(yīng)數(shù)據(jù)表的字段屬性, 因?yàn)閍ttrs中包含所有的類屬性
            # 這里就簡(jiǎn)單判斷字段是元組
            if isinstance(v, tuple):
                print('Found mapping: %s ==> %s' % (k, v))
                mappings[k] = v

        # 刪除這些已經(jīng)在字典中存儲(chǔ)的字段屬性
        for k in mappings.keys():
            attrs.pop(k)

        # 將之前的uid/name/email/password以及對(duì)應(yīng)的對(duì)象引用、類名字
        # 用其他類屬性名稱保存
        attrs['__mappings__'] = mappings  # 保存屬性和列的映射關(guān)系
        attrs['__table__'] = cls_name  # 假設(shè)表名和類名一致
        return type.__new__(mcs, cls_name, bases, attrs)


class Model(object, metaclass=ModelMetaclass):
    """數(shù)據(jù)表模型基類"""

    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))

        # 把參數(shù)數(shù)據(jù)類型對(duì)應(yīng)數(shù)據(jù)表的字段類型
        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__
        # 數(shù)據(jù)表中的字段
        fields = ','.join(fields)
        # 待插入的數(shù)據(jù)
        args = ','.join(args_temp)

        # 生成sql語(yǔ)句
        sql = f"""insert into {table_name} ({fields}) values ({args})"""
        print(f'SQL: {sql}')

        # 執(zhí)行sql語(yǔ)句
        # ...


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()

添加數(shù)據(jù)庫(kù)驅(qū)動(dòng)執(zhí)行sql語(yǔ)句

這里我們使用 pymysql 數(shù)據(jù)庫(kù)驅(qū)動(dòng)，來(lái)執(zhí)行 sql 語(yǔ)句

在 Model 類中新增一個(gè) get_connection 的靜態(tài)方法用于獲取數(shù)據(jù)庫(kù)連接

import pymysql


class Model(object, metaclass=ModelMetaclass):
    """數(shù)據(jù)表模型基類"""

    def __init__(self, **kwargs):
        for name, value in kwargs.items():
            setattr(self, name, value)

    @staticmethod
    def get_connection():
        """
        獲取數(shù)據(jù)庫(kù)連接與數(shù)據(jù)游標(biāo)
        :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))

        # 把參數(shù)數(shù)據(jù)類型對(duì)應(yīng)數(shù)據(jù)表的字段類型
        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__
        # 數(shù)據(jù)表中的字段
        fields = ','.join(fields)
        # 待插入的數(shù)據(jù)
        args = ','.join(args_temp)

        # 生成sql語(yǔ)句
        sql = f"""insert into {table_name} ({fields}) values ({args})"""
        print(f'SQL: {sql}')

        # 執(zhí)行sql語(yǔ)句
        conn, cursor = self.get_connection()
        ret = cursor.execute(sql)
        print(ret)
        conn.commit()
        cursor.close()
        conn.close()

添加數(shù)據(jù)庫(kù)驅(qū)動(dòng)執(zhí)行sql語(yǔ)句

這里我們使用 pymysql 數(shù)據(jù)庫(kù)驅(qū)動(dòng)，來(lái)執(zhí)行 sql 語(yǔ)句

在 Model 類中新增一個(gè) get_connection 的靜態(tài)方法用于獲取數(shù)據(jù)庫(kù)連接

import pymysql


class Model(object, metaclass=ModelMetaclass):
    """數(shù)據(jù)表模型基類"""

    def __init__(self, **kwargs):
        for name, value in kwargs.items():
            setattr(self, name, value)

    @staticmethod
    def get_connection():
        """
        獲取數(shù)據(jù)庫(kù)連接與數(shù)據(jù)游標(biāo)
        :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))

        # 把參數(shù)數(shù)據(jù)類型對(duì)應(yīng)數(shù)據(jù)表的字段類型
        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__
        # 數(shù)據(jù)表中的字段
        fields = ','.join(fields)
        # 待插入的數(shù)據(jù)
        args = ','.join(args_temp)

        # 生成sql語(yǔ)句
        sql = f"""insert into {table_name} ({fields}) values ({args})"""
        print(f'SQL: {sql}')

        # 執(zhí)行sql語(yǔ)句
        conn, cursor = self.get_connection()
        ret = cursor.execute(sql)
        print(ret)
        conn.commit()
        cursor.close()
        conn.close()

測(cè)試功能

準(zhǔn)備數(shù)據(jù)庫(kù)

先準(zhǔn)備數(shù)據(jù)庫(kù) testdb 和 user 數(shù)據(jù)表

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 表結(jié)構(gòu)如下

+----------+------------------+------+-----+---------+----------------+
| 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    |                |
+----------+------------------+------+-----+---------+----------------+

創(chuàng)建模型類測(cè)試

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()

查看數(shù)據(jù)庫(kù) user 表數(shù)據(jù)

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)

以上就是ORM框架怎么實(shí)現(xiàn)的全部?jī)?nèi)容，更多Python學(xué)習(xí)資料請(qǐng)關(guān)注W3Cschool其它相關(guān)文章！


Python

0 人點(diǎn)贊