手撸ORM

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了手撸ORM相关的知识,希望对你有一定的参考价值。

 
ORM即Object Relational Mapping,全称对象关系映射。
当我们需要对数据库进行操作时,势必需要通过连接数据、调用sql语句、执行sql语句等操作,ORM将数据库中的表,字段,行与我们面向对象编程的类及其方法,属性等一一对应,即将该部分操作封装起来,程序猿不需懂得sql语句即可完成对数据库

Django\'s ORM 
优点:
1. 易用,学习曲线短 
2. 和Django紧密集合,用Django时使用约定俗成的方法去操作数据库 
缺点:
3. 不好处理复杂的查询,强制开发者回到原生SQL 
4. 紧密和Django集成,使得在Django环境外很难使用 
 
peewee 
优点:
5. Django式的API,使其易用 
6. 轻量实现,很容易和任意web框架集成 
缺点:
7. 多对多查询写起来不直观 
 
SQLAlchemy 
优点:
8. 企业级 API,使得代码有健壮性和适应性 
9. 灵活的设计,使得能轻松写复杂查询 
缺点:
10. 重量级 API,导致长学习曲线 
其它:SQLObject 、Storm 、、、、
 
ORM 池版
Db_pool
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
)

fuckorm

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)

Mysql_singleton

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

ORM 单列版

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)

 

以上是关于手撸ORM的主要内容,如果未能解决你的问题,请参考以下文章

手撸orm

手撸ORM

手撸系列之——ORM(对象关系映射)

很多小伙伴不太了解ORM框架的底层原理,这不,冰河带你10分钟手撸一个极简版ORM框架(赶快收藏吧)

很多小伙伴不太了解ORM框架的底层原理,这不,冰河带你10分钟手撸一个极简版ORM框架(赶快收藏吧)

02 django 框架ORM 操作数据库,字段的增删改查