Skip to content

Latest commit

 

History

History
429 lines (290 loc) · 10.5 KB

4.元类编写orm.md

File metadata and controls

429 lines (290 loc) · 10.5 KB

元类实现ORM

1. ORM是什么

ORM 是 python编程语言后端web框架 Django的核心思想，“Object Relational Mapping”，即对象-关系映射，简称ORM。

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

demo:

class User(父类省略):
    uid = ('uid', "int unsigned")
    name = ('username', "varchar(30)")
    email = ('email', "varchar(30)")
    password = ('password', "varchar(30)")
    ...省略...


u = User(uid=12345, name='Michael', email='test@orm.org', password='my-pwd')
u.save()
# 对应如下sql语句
# insert into User (username,email,password,uid)
# values ('Michael','test@orm.org','my-pwd',12345)

说明

  1. 所谓的ORM就是让开发者在操作数据库的时候，能够像操作对象时通过xxxx.属性=yyyy一样简单，这是开发ORM的初衷
  2. 只不过ORM的实现较为复杂，Django中已经实现了 很复杂的操作，本节知识 主要通过完成一个 insert相类似的ORM，理解其中的道理就就可以了

2.普通的sql写法

# -*- coding:utf-8 -*-
from pymysql import *


def create():
    create_dict = {"uid": "int unsigned", "name": "varchar(30)", "email": "varchar(30)", "password": "varchar(30)"}

    # 创建Connection连接
    conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
    # 获得Cursor对象
    cs1 = conn.cursor()

    fields = list()

    for key, value in create_dict.items():
        fields.append("%s %s" % (key, value))

    # 创建表
    # create_sql = """ CREATE TABLE IF NOT EXISTS user(uid int unsigned,name varchar(30),email varchar(30),password varchar(30));"""
    create_sql = """ CREATE TABLE IF NOT EXISTS user(%s);""" % (",".join(fields),)

    print(create_sql)

    cs1.execute(create_sql)

    # 提交
    conn.commit()

    # 关闭
    cs1.close()
    conn.close()


def insert(**kwargs):
    print(kwargs)
    # 创建Connection连接
    conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
    # 获得Cursor对象
    cs1 = conn.cursor()

    # 插入数据
    # insert_sql = """ insert into user (uid,name,email,password) values (123,'oldyang','test@orm.org','pwd');"""
    keys = list()
    values = list()

    for key, value in kwargs.items():
        keys.append(key)
        # 如果是Int转成字符串
        if isinstance(value, int):
            # 说明 int
            values.append(str(value))
        else:
            values.append(""" "%s" """ % value)

    insert_sql = """ insert into user (%s) values (%s);""" % (",".join(keys), ",".join(values))

    print(insert_sql)

    cs1.execute(insert_sql)

    # 提交
    conn.commit()

    # 关闭
    cs1.close()
    conn.close()


def main():
    create()

    insert(uid=123, password='pwd', email='test@123.com', name='toby')


if __name__ == '__main__':
    main()

3.使用元类封装

# -*- coding:utf-8 -*-
# -*- coding:utf-8 -*-
from pymysql import *


class MyMetaClass(type):
    """元类生成字典"""

    def __new__(cls, class_name, supers_name, attrs):
        create_dict = dict()  # 创建一个新的字典来存放想要的数据
        for key, value in attrs.items():
            if isinstance(value, tuple):
                create_dict[key] = value[0]
        attrs["create_dict"] = create_dict  # 添加新的类属性
        return type.__new__(cls, class_name, supers_name, attrs)


class User(object, metaclass=MyMetaClass):
    uid = ("int unsigned",)
    name = ("varchar(30)",)
    email = ("varchar(30)",)
    password = ("varchar(30)",)

    def create(self):
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        fields = list()

        for key, value in self.create_dict.items():
            fields.append("%s %s" % (key, value))

        # 创建表
        create_sql = """ CREATE TABLE IF NOT EXISTS user(%s);""" % (",".join(fields),)

        print(create_sql)

        cs1.execute(create_sql)

        # 提交
        conn.commit()

        # 关闭
        cs1.close()
        conn.close()

    def insert(self, **kwargs):
        print(kwargs)
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        # 插入数据
        keys = list()
        values = list()

        for key, value in kwargs.items():
            keys.append(key)
            # 如果是Int转成字符串
            if isinstance(value, int):
                # 说明 int
                values.append(str(value))
            else:
                values.append(""" "%s" """ % value)

        insert_sql = """ insert into user (%s) values (%s);""" % (",".join(keys), ",".join(values))

        print(insert_sql)

        cs1.execute(insert_sql)

        # 提交
        conn.commit()

        # 关闭a
        cs1.close()
        conn.close()


def main():
    user = User()
    user.create()
    user.insert(uid=123, password='pwd', email='test@123.com', name='toby')


if __name__ == '__main__':
    main()

4.继承封装且类名就是数据库名

# -*- coding:utf-8 -*-
# -*- coding:utf-8 -*-
from pymysql import *


class MyMetaClass(type):
    """元类生成字典"""

    def __new__(cls, class_name, supers_name, attrs):
        create_dict = dict()  # 创建一个新的字典来存放想要的数据

        for key, value in attrs.items():
            if isinstance(value, tuple):
                create_dict[key] = value[0]
        attrs["table_name"] = class_name.lower()  # 添加表名
        attrs["create_dict"] = create_dict  # 添加新的类属性

        return type.__new__(cls, class_name, supers_name, attrs)


class Table(object, metaclass=MyMetaClass):
    """创建表的父类"""

    def create(self):
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        fields = list()

        for key, value in self.create_dict.items():
            fields.append("%s %s" % (key, value))

        # 创建表
        create_sql = """ CREATE TABLE IF NOT EXISTS %s(%s);""" % (self.table_name, ",".join(fields),)

        print(create_sql)

        cs1.execute(create_sql)

        # 提交
        conn.commit()

        # 关闭
        cs1.close()
        conn.close()

    def insert(self, **kwargs):
        print(kwargs)
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        # 插入数据
        keys = list()
        values = list()

        for key, value in kwargs.items():
            keys.append(key)
            # 如果是Int转成字符串
            if isinstance(value, int):
                # 说明 int
                values.append(str(value))
            else:
                values.append(""" "%s" """ % value)

        insert_sql = """ insert into user (%s) values (%s);""" % (",".join(keys), ",".join(values))

        print(insert_sql)

        cs1.execute(insert_sql)

        # 提交
        conn.commit()

        # 关闭a
        cs1.close()
        conn.close()


class User(Table):
    """继承来写"""
    uid = ("int unsigned",)
    name = ("varchar(30)",)
    email = ("varchar(30)",)
    password = ("varchar(30)",)


def main():
    user = User()
    user.create()
    user.insert(uid=123, password='pwd', email='test@123.com', name='toby')


if __name__ == '__main__':
    main()

5.属性换成类处理

# -*- coding:utf-8 -*-
from pymysql import *


class Field(object):
    def __init__(self, data):
        self.data = data


class MyMetaClass(type):
    """元类生成字典"""

    def __new__(cls, class_name, supers_name, attrs):
        create_dict = dict()  # 创建一个新的字典来存放想要的数据

        for key, value in attrs.items():
            if isinstance(value, Field):

                create_dict[key] = value.data

        attrs["table_name"] = class_name.lower()  # 添加表名
        attrs["create_dict"] = create_dict  # 添加新的类属性

        return type.__new__(cls, class_name, supers_name, attrs)


class Table(object, metaclass=MyMetaClass):
    """创建表的父类"""

    def create(self):
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        fields = list()

        for key, value in self.create_dict.items():
            fields.append("%s %s" % (key, value))

        # 创建表
        create_sql = """ CREATE TABLE IF NOT EXISTS %s(%s);""" % (self.table_name, ",".join(fields),)

        print(create_sql)

        cs1.execute(create_sql)

        # 提交
        conn.commit()

        # 关闭
        cs1.close()
        conn.close()

    def insert(self, **kwargs):
        print(kwargs)
        # 创建Connection连接
        conn = connect(host='localhost', port=3306, database='stock_db', user='root', password='mysql', charset='utf8')
        # 获得Cursor对象
        cs1 = conn.cursor()

        # 插入数据
        keys = list()
        values = list()

        for key, value in kwargs.items():
            keys.append(key)
            # 如果是Int转成字符串
            if isinstance(value, int):
                # 说明 int
                values.append(str(value))
            else:
                values.append(""" "%s" """ % value)

        insert_sql = """ insert into user (%s) values (%s);""" % (",".join(keys), ",".join(values))

        print(insert_sql)

        cs1.execute(insert_sql)

        # 提交
        conn.commit()

        # 关闭a
        cs1.close()
        conn.close()


class User(Table):
    """继承来写"""
    uid = Field("int unsigned", )
    name = Field("varchar(30)", )
    email = Field("varchar(30)", )
    password = Field("varchar(30)", )


def main():
    user = User()
    user.create()
    user.insert(uid=123, password='pwd', email='test@123.com', name='toby')


if __name__ == '__main__':
    main()