手撸ORM

from DBUtils.PooledDB import PooledDB
import pymysql
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=[],  # 开始会话前执行的命令列表。
    ping=0,
    # ping MySQL服务端，检查是否服务可用。
    host=‘127.0.0.1‘,
    port=3306,
    user=‘root‘,
    password=‘123‘,
    database=‘youku‘,
    charset=‘utf8‘,
    autocommit=True
)

import pymysql
from conf import setting
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=6,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
    maxcached=5,  # 链接池中最多闲置的链接，0和None不限制
    maxshared=3,
    # 链接池中最多共享的链接数量，0和None表示全部共享。
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
    setsession=[],  # 开始会话前执行的命令列表。
    ping=0,
    # ping MySQL服务端，检查是否服务可用。

    host=setting.host,
    port=setting.port,
    user=setting.user,
    password=setting.password,
    database=setting.database,
    charset=setting.charset,
    autocommit=setting.autocommit
)

from youkuServer.orm_pool import Mysql_singleton

class Filepd:
    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 Setclass(Filepd):
    def __init__(self,name=None,column_type=‘varchar(200)‘,primary_key=None,default=None):
        super().__init__(name,column_type,primary_key,default)

class Intclass(Filepd):
    def __init__(self,name=None,column_type=‘int‘,primary_key=None,default=0):
        super().__init__(name,column_type,primary_key,default)

class Modledsclass(type):
    def __new__(cls, name,based,attrs):
        if name==‘Modles‘:
            return type.__new__(cls,name,based,attrs)
        table_name=attrs.get(‘table_name‘,None)

        primary_key=None
        mappings=dict()
        for k,v in attrs.items():
            if isinstance(v,Filepd):
                mappings[k]=v
                if v.primary_key:

                    if primary_key:
                        raise TypeError(‘主键存在：%s‘%k)
                    primary_key=k
        for k in mappings.keys():
            attrs.pop(k)
        if not primary_key:
            raise TypeError(‘没有主键‘)
        attrs[‘table_name‘]=table_name
        attrs[‘primary_key‘]=primary_key
        attrs[‘mappings‘]=mappings
        return type.__new__(cls,name,based,attrs)

class Modles(dict,metaclass=Modledsclass):
    def __init__(self,**kwargs):
        super().__init__(**kwargs)
    def __setattr__(self, key, value):
        self[key]=value

    def __getattr__(self, item):
        try:
            return self[item]
        except TypeError:
            raise (‘没有该属性‘)

    @classmethod
    def select_ont(cls,**kwargs):
        ms=Mysql_singleton.Mysql()
        key=list(kwargs.keys())[0]
        value=kwargs[key]

        sql=‘select * from %s where %s=? ‘%(cls.table_name,key)
        sql=sql.replace(‘?‘,‘%s‘)
        re=ms.select(sql,value)
        if re:
            u=cls(**re[0])
            return u
        else:
            return
    @classmethod
    def select_mary(cls,**kwargs):
        ms=Mysql_singleton.Mysql()
        if kwargs:
            key=list(kwargs.keys())[0]
            value=kwargs[key]

            sql=‘select * from %s where %s=?‘%(cls.table_name,key)
            sql=sql.replace(‘?‘,‘%s‘)
            re = ms.select(sql, value)
        else:
            sql = ‘select * from %s‘ % (cls.table_name)
            re = ms.select(sql)
        if re:
            lis_obj = [cls(**r) for r in re]
            return lis_obj
        else:
            return

    def update(self):
        ms=Mysql_singleton.Mysql()
        filde=[]
        pr=None
        args=[]
        for k,v in self.mappings.items():
            if v.primary_key:
                pr=getattr(self,v.name,None)
                filde.append(v.name + ‘=?‘)
                args.append(getattr(self,v.name,v.default))

        sql=‘update %s set % where %s=%s‘%(self.table_name,‘,‘.join(filde),self.primary_key,pr)
        sql=sql.replace(‘?‘,‘%s‘)
        ms.select(sql,args)
    def save(self):
        ms=Mysql_singleton.Mysql()
        filde=[]
        values=[]
        args=[]
        for k,v in self.mappings.items():
            if not v.primary_key:
                filde.append(v.name)
            else:
                values.append(‘?‘)
                args.append(getattr(self,v.name,v.default))
        sql=‘insert into %s (%s) values (%s)‘%(self.table_name,‘,‘.join(filde),‘,‘.join(values))
        sql=sql.replace(‘?‘,‘%s‘)
        ms.select(sql,args)

import pymysql
from youkuServer.orm_pool import db_pool

class Mysql:
    def __init__(self):
        self.conn=db_pool.POOL.connection()
        self.cursor=self.conn.cursor(cursor=pymysql.cursors.DictCursor)

    def close(self):
        self.conn.close()
        self.cursor.close()

    def select(self,sql,args=None):
        self.cursor.execute(sql,args)
        rs=self.cursor.fetchall()
        return rs

    def execute(self,sql,args):
        try:
            self.cursor.execte(sql,args)
            aff=self.cursor.rowcount
        except BaseException as e:
            print(e)
        return aff

from youkuServer.orm import Mysql_singleton

class Fileld:
    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 StringFileld(Fileld):
    def __init__(self,name=None,column_type=‘varchar(200)‘,primary_key=False,default=None):
        super().__init__(name,column_type,primary_key,default)



class IntegerFileld(Fileld):
    def __init__(self,name=None,column_type=‘int‘,primary_key=False,default=0):
        super().__init__(name,column_type,primary_key,default)


class ModlesMetaclass(type):
    def __new__(cls, name,bases,attrs):

        if name==‘Modles‘:
            return type.__new__(cls,name,bases,attrs)
        table_name=attrs.get(‘table_name‘,None)
        # table_name=attrs[‘table_name‘]

        primary_key=None
        mappings=dict()
        for k,v in attrs.items():
            if isinstance(v,Fileld):#v 是不是Field的对象
                mappings[k]=v
                if v.primary_key:

                    #找到主键
                    if primary_key:
                        raise TypeError(‘主键重复：%s‘%k)
                    primary_key=k

        for k in mappings.keys():
            attrs.pop(k)
        if not primary_key:
            raise TypeError(‘没有主键‘)
        attrs[‘table_name‘]=table_name
        attrs[‘primary_key‘]=primary_key
        attrs[‘mappings‘]=mappings
        return type.__new__(cls,name,bases,attrs)









class Modles(dict,metaclass=ModlesMetaclass):
    def __init__(self,**kwargs):
        super().__init__(**kwargs)


    def __setattr__(self, key, value):

        self[key]=value

    def __getattr__(self, item):
        try:
          return self[item]
        except TypeError:
            raise (‘没有该属性‘)

    @classmethod
    def select_one(cls,**kwargs):
        #只查一条
        key=list(kwargs.keys())[0]
        value=kwargs[key]
        #select * from user where id=%s
        sql=‘select * from %s where %s=?‘%(cls.table_name,key)
        #
        sql=sql.replace(‘?‘,‘%s‘)
        ms=Mysql_singleton.Mysql().singleton()
        re=ms.select(sql,value)
        if re:
            #attrs={‘name‘:‘123‘,‘password‘:123}
            #u=User(**attrs)
            #相当于 User(name=‘123‘,password=123)
            u=cls(**re[0])
            return u
        else:
            return

    @classmethod
    def select_many(cls,**kwargs):
        ms = Mysql_singleton.Mysql().singleton()
        if kwargs:
            key=list(kwargs.keys())[0]
            value=kwargs[key]
            sql = ‘select * from %s where %s=?‘ % (cls.table_name, key)
            #
            sql = sql.replace(‘?‘, ‘%s‘)

            re = ms.select(sql, value)
        else:
            sql = ‘select * from %s‘%(cls.table_name)
            re = ms.select(sql)

        if re:
            lis_obj=[cls(**r) for r in re]
            return lis_obj
        else:
            return

    def update(self):
        ms = Mysql_singleton.Mysql().singleton()
        #update user set name=?,password=? where id=1

        filed=[]
        pr=None
        args=[]
        for k,v in self.mappings.items():

            if v.primary_key:
                pr=getattr(self,v.name,None)
            else:
                filed.append(v.name + ‘=?‘)
                args.append(getattr(self,v.name,v.default))

        sql = ‘update %s set %s where %s =%s‘%(self.table_name,‘,‘.join(filed),self.primary_key,pr)
        #‘update user set name=?,password =? where id =1‘
        sql=sql.replace(‘?‘,‘%s‘)
        ms.execute(sql,args)
    def save(self):
        ms = Mysql_singleton.Mysql().singleton()
        #insert into user (name,passwword) values (?,?)
        filed=[]
        values=[]
        args=[]
        for k,v in self.mappings.items():
            if not v.primary_key:
                filed.append(v.name)
                values.append(‘?‘)
                args.append(getattr(self,v.name,v.default))
        sql =‘insert into %s (%s) VALUES (%s)‘%(self.table_name,‘,‘.join(filed),‘,‘.join(values))
        sql= sql.replace(‘?‘,‘%s‘)
        ms.execute(sql,args)

import pymysql

class Mysql:
    __instense=None
    def __init__(self):
        self.conn=pymysql.connect(
            host=‘127.0.0.1‘,
            port=3306,
            user=‘root‘,
            password=‘123‘,
            charset=‘utf8‘,
            database=‘youku‘,
            autocommit=True
        )
        self.cursor=self.conn.cursor(cursor=pymysql.cursors.DictCursor)

    def close_db(self):
        self.cursor.close()
        self.conn.close()

    def select(self, sql, args=None):
        #select * from user where id=%s

        self.cursor.execute(sql, args)
        rs = self.cursor.fetchall()
        return rs

    def execute(self, sql, args):
        try:
            #update user set name=‘oo‘ where id =%s
            self.cursor.execute(sql, args)
            affected = self.cursor.rowcount
            # self.conn.commit()
        except BaseException as e:
            print(e)
        return affected
    @classmethod
    def singleton(cls):
        if not cls.__instense:

            cls.__instense=cls()

        return  cls.__instense




if __name__ == ‘__main__‘:
    ms=Mysql()
    re=ms.select(‘select * from user where id=%s‘,1)
    print(re)