手写版本orm

orm.py

'''
定义字段类
'''
from orm.mysql_control import Mysql


class Field:
    def __init__(self, name, column_type, primary_key, default):
        self.name = name
        self.column_type = column_type
        self.primary_key = primary_key
        self.default = default


# varchar
class StringField(Field):
    def __init__(self, name, column_type='varchar(255)', primary_key=False, default=None):
        super().__init__(name, column_type, primary_key, default)


# int
class IntegerField(Field):
    def __init__(self, name, column_type='int', primary_key=False, default=0):
        super().__init__(name, column_type, primary_key, default)


# 元类控制表模型类的创建
class OrmMetaClass(type):

    # 类名, 类的基类, 类的名称空间
    def __new__(cls, class_name, class_bases, class_attr):
        # print(class_name, class_bases, class_attr)
        # 1.过滤Models类
        if class_name == 'Models':
            return type.__new__(cls, class_name, class_bases, class_attr)

        # 2.控制模型表中: 表名, 主键, 表的字段
        # 如果模型表类中没有定义table_name,把类名当做表名

        # 获取表名
        table_name = class_attr.get('table_name', class_name)  # user_info, User

        # 3.判断是否只有一个主键
        primary_key = None

        # 用来存放所有的表字段, 存不是目的,目的是为了取值方便
        mappings = {}

        '''
        __main__: xxxx
        'id': <__main__.IntegerField object at 0x000001E067D48B00>, 
        'name': <__main__.StringField object at 0x000001E067D48AC8>}
        '''
        for key, value in class_attr.items():

            # 判断value是否是字段类的对象
            if isinstance(value, Field):

                # 把所有字段都添加到mappings中
                mappings[key] = value

                if value.primary_key:

                    if primary_key:
                        raise TypeError('主键只能有一个')

                    # 获取主键
                    primary_key = value.name

        # 删除class_attr中与mappings重复的属性, 节省资源
        for key in mappings.keys():
            class_attr.pop(key)

        # 判断是否有主键
        if not primary_key:
            raise TypeError('必须要有一个主键')

        class_attr['table_name'] = table_name
        class_attr['primary_key'] = primary_key
        class_attr['mappings'] = mappings
        '''
               'table_name': table_name
               'primary_key': primary_key
               'mappings': {'id': <__main__.IntegerField object at 0x000001E067D48B00>,
                            'name': <__main__.StringField object at 0x000001E067D48AC8>}
                            }
       '''
        return type.__new__(cls, class_name, class_bases, class_attr)


# 继承字典类,
class Models(dict, metaclass=OrmMetaClass):
    def __init__(self, **kwargs):
        # print(kwargs)  # 接收关键字参数
        super().__init__(**kwargs)

    # 在对象.属性没有的时候触发
    def __getattr__(self, item):
        # print(item)
        return self.get(item, '没有这个key')

    # 在对象.属性 = 属性值 时触发
    def __setattr__(self, key, value):

        # 字典赋值操作
        self[key] = value

    # 查
    @classmethod
    def select(cls, **kwargs):

        # 获取数据库链接对象
        ms = Mysql()

        # 若没有kwargs代表没有条件查询
        if not kwargs:
            # select * from table;
            sql = 'select * from %s' % cls.table_name

            res = ms.my_select(sql)

        # 若有kwargs代表有条件
        else:
            # print(kwargs)  # {id:1}
            key = list(kwargs.keys())[0]  # id
            value = kwargs.get(key)  # 1

            # select * from table where id=1;
            sql = 'select * from %s where %s=?' % (
                cls.table_name, key
            )

            sql = sql.replace('?', '%s')

            res = ms.my_select(sql, value)

        if res:
            # [{},{}, {}]   ---->  [obj1, obj2, obj3]
            # 把mysql返回来的 列表套 字典 ---> 列表套 对象
            # l1 = []
            # # 遍历mysql返回所有的字典
            # for d in res:
            #     # 把每一个字典传给cls实例化成一个个的r1对象
            #     r1 = cls(**d)
            #     # 追加到l1列表中
            #     l1.append(r1)

            return [cls(**result) for result in res]

    # 插入
    def save(self):
        ms = Mysql()
        # insert into table(x,x,x) values(x,x,x);

        # 字段名
        fields = []
        # 字段的值
        values = []
        # 存放对应字段的?号
        args = []

        for k, v in self.mappings.items():
            # 把主键过滤掉
            if not v.primary_key:
                fields.append(
                    v.name
                )
                values.append(
                    getattr(self, v.name, v.default)
                )
                args.append('?')

        # insert into table(x,x,x) values(?, ?, ?);
        sql = 'insert into %s(%s) values(%s)' % (
            self.table_name, ','.join(fields), ','.join(args)
        )

        sql = sql.replace('?', '%s')

        ms.my_execute(sql, values)

    # 更新
    def sql_update(self):
        ms = Mysql()

        fields = []
        primary_key = None
        values = []

        for k, v in self.mappings.items():
            # 获取主键的值
            if v.primary_key:
                primary_key = getattr(self, v.name, v.default)

            else:

                # 获取 字段名=?, 字段名=?,字段名=?
                fields.append(
                    v.name + '=?'
                )

                # 获取所有字段的值
                values.append(
                    getattr(self, v.name, v.default)
                )

        # update table set %s=?,... where id=1;  把主键当做where条件
        sql = 'update %s set %s where %s=%s' % (
            self.table_name, ','.join(fields), self.primary_key, primary_key
        )

        # print(sql)  # update User set name=? where id=3

        sql = sql.replace('?', '%s')

        ms.my_execute(sql, values)


# User, Movie, Notice
# 表模型类
class User(Models):
    # table_name = 'user_info'
    id = IntegerField(name='id', primary_key=True)
    name = StringField(name='name')
    # pwd = StringField(name='pwd')


class Movie(Models):
    id = IntegerField(name='id', primary_key=True)
    pass


# # # User('出入任意个数的关键字参数')
# user_obj = User()  # user_obj--->dict
# user_obj.name = 'xxxx'

# if __name__ == '__main__':
#     res = User.select(name='jason_sb')[0]
#     print(res)
#
#     # res.name = 'jason_sb'
#     #
#     # res.sql_update()
#
#     # user_obj = User(name='egon')
#     # user_obj.save()
#
# '''
# 表:
#     表名, 只有一个唯一的主键, 字段(必须是Field的字段)
#
# 元类:
#     通过元类控制类的创建.
# '''
#
# # class Movie:
# #     def __init__(self, movie_name, movie_type):
# #         self.movie_name = movie_name
# #         self.movie_type = movie_type
# #
# #
# # class Notice:
# #     def __init__(self, title, content):
# #         self.title = title
# #         self.content = content
#
# '''
# 问题1: 所有表类都要写__init__, 继承一个父类
# 问题2: 可以接收任意个数以及任意名字的关键字参数. 继承python中的字典对象.
# '''
#
# # if __name__ == '__main__':
# #     # d1 = dict({'name': 'tank'})
# #     # d2 = dict(name='tank2')
# #     # print(d1)
# #     # print(d2)
# #
# #     d3 = Models(name='jason')
# #     # print(d3)
# #     # print(d3.get('name'))
# #     # print(d3['name'])
# #     # print(d3.name)
# #     # d3.name = 'tank'
# #     # d3.pwd = '123'
# #     # print(d3.name)
# #     # print(d3)
# #     print(d3.name)  # None
# #
# #     d3.pwd = '123'
# #     print(d3.pwd)
# #     print(d3)


if __name__ == '__main__':
    # 查看所有
    # res = User.select()
    # print(res)

    # 根据查询条件查询
    res = User.select(name='json_egon_sb')
    print(res)
#
#     # 更新
#     # user_obj = res[0]
#     # user_obj.name = 'jason_sb_sb'
#     # user_obj.sql_update()  # {'id': 3, 'name': 'jason_sb'}
#
#     # 插入
#     # user_obj = User(name='json_egon_sb')
# #     # user_obj.save()

mysql_control.py

import pymysql
from orm.db_pool import POOL


class Mysql:

    def __init__(self):
        # 建立链接
        self.conn = POOL.connection()

        # 获取游标
        self.cursor = self.conn.cursor(
            pymysql.cursors.DictCursor
        )

    # 关闭游标链接方法
    def close_db(self):
        self.cursor.close()
        self.conn.close()

    # 查看
    def my_select(self, sql, args=None):

        self.cursor.execute(sql, args)

        res = self.cursor.fetchall()
        # [{}, {}, {}]
        # print(res)
        return res

    # 提交
    def my_execute(self, sql, args):
        try:
            # 把insert , update...一系列sql提交到mysql中
            self.cursor.execute(sql, args)

        except Exception as e:
            print(e)

db_pool.py

from DBUtils.PooledDB import PooledDB
import pymysql
# pip3 install DBUtils

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
    maxcached=5,  # 链接池中最多闲置的链接，0和None不限制
    maxshared=3,
    # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
    setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
    ping=0,
    # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='youku_demo',
    charset='utf8',
    autocommit='True'
)