博客整理 day39 ORM系统实现

目录

• python day 39 ORM系统实现
  • 一 字段类
  • 二 自定义元类
  • 三 Models类
  • 四 mysql连接类

python day 39 ORM系统实现

一 字段类

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

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

class StringField(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_base, class_dict):
        # print(class_name, class_base,class_dict)
        if class_name == 'Models':
            return type.__new__(cls, class_name, class_base, class_dict)

        table_name = class_dict.get('table_name', class_name)

        primary_key = None
        mappings = {}

        for key, value in class_dict.items():
            # print(key, value)
            if isinstance(value, Field):
                mappings[key] = value

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

                    primary_key = value.name

        for key in mappings.keys():
            class_dict.pop(key)

        if not primary_key:
            raise TypeError('必需要有一个主键')

        class_dict['table_name'] = table_name
        class_dict['primary_key'] = primary_key
        class_dict['mappings'] = mappings

        return type.__new__(cls, class_name, class_base, class_dict)

三 Models类

class Models(dict, metaclass=OrmMetaClass):
    def __setattr__(self, key, value):
        self[key] = value

    def __getattr__(self, item):
        return self.get(item)

    @classmethod
    def orm_select(cls, **kwargs):
        mysql = MySQLClient()
        if not kwargs:
            sql = 'select * from %s' % cls.table_name
            res = mysql.sql_select(sql)
        else:
            key = list(kwargs.keys())[0]
            value = kwargs.get(key)

        sql = 'select * from %s where %s = ?' %(cls.table_name,key)
        sql = sql.replace('?','%s')
        res = mysql.sql_select(sql,value)

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

    def orm_insert(self):
        mysql = MySQLClient()
        keys = []
        values = []
        args = []

        for key,value in self.mappings.items():
            if not value.primary_key:
                keys.append(key)
                values.append(
                    getattr(self,value.name,value.default)
                )

                args.append('?')

        sql = 'insert into %s (%s) values (%s)' %(
            self.table_name,
            ','.join(keys),
            ','.join(args)
        )
        sql = sql.replace('?','%s')

        mysql.sql_execute(sql,values)

    def orm_update(self):
        mysql = MySQLClient()
        keys = []
        values = []
        primary_key = None

        for key,value in self.mappings.items():
            if value.primary_key:
                primary_key = value.name + '=%s' %getattr(self,value.name)
            else:
                keys.append(value.name+'=?')
                values.append(
                    getattr(self,value.name)
                )

        sql = 'update %s set %s where %s' %(
            self.table_name,
            ','.join(keys),
            primary_key
        )
        sql = sql.replace('?','%s')
        mysql.sql_execute(sql,values)

四 mysql连接类

import pymysql

class MySQLClient:

    def __init__(self):
        self.client = pymysql.connect(
            host='localhost',
            user='root',
            password='123',
            database='ormtest',
            charset='utf8',
            autocommit=True
        )
        self.cursor = self.client.cursor(pymysql.cursors.DictCursor)

    def sql_select(self, sql, value=None):
        self.cursor.execute(sql, value)
        res = self.cursor.fetchall()
        return res

    def sql_execute(self,sql, values):
        try:
            self.cursor.execute(sql, values)
        except Exception as e:
            print(e)