orm之sqlarchemy 底层实现原理

Posted 2022-01-24 qingqinxu

ORM 【即Object Relational Mapping，对象关系映射】sqlarchemy底层是如何实现的?
当我们需要对数据库进行操作时，是否可以依如下方式操作数据库-- 不要写sql语句，我要写python代码

创建表   --------   创建类

行数据   --------   对象

操作行数据  ----  操作对象   

问题：1.首先需要一个类，实例一个对象就是一行数据。

　　   2. 是否可以找到一种用点的方式的获取数据的数据类型，因为给对象添加属性：【obj.name=‘egon‘】 ，访问对象属性【obj.name】

　　      所学数据类型中貌似字典比较可以，先来实例化一个字典对象

　　  dic = dict(name=‘egon‘,age=18)
　　  print(dic)  #‘name‘: ‘egon‘, ‘age‘: 18
　　　　        

　　　3.  字典只能加中括号 dic[ ‘ name ‘] 取值，赋值。是否可以重写一个dict 类，改变一下对象dic取值的方式？

　　　　 可通过重写dict类中的内置方法__ setattr__、 __getattr__ 使字典对象可以像普通对象一样用给自己添加属性【obj.name=‘ jerry ‘】、访问属性【obj.name】。

　　　　class A(dict):

   　　　　　　 def __setattr__(self, key, value):
        　　　　　　#修改、新增时触发
         　　　　　　print(‘这是settattr key:%a, value:%s‘%(key,value))
       　　　　　　  self[key] = value

    　　　　　　def __getattr__(self, item):
       　　　　　　 print(‘getattr item:%s‘% item)
        　　　　　　#取值时触发
       　　　　　　 try:
          　　　　　　 return self[item]
       　　　　　　 except TypeError:
           　　　　　　　　 raise (‘没有该属性 ‘)
　　　　obj = A()

　　　　# obj[‘name‘]=‘egon‘     原来就有的添加属性的方式
　　　　# obj[‘password‘] =123   
　　　　# 

　　　　obj.name = ‘jerry‘  # 新增、修改时会触发__setattr__ 的执行
　　　　print(obj.name)  # jerry  对象取值时触发__getattr__ 的执行

　　　　4. 对象添加、获取数据用点方法问题解决了之后，需要对类的实例进行限制：

　　　　　　a. 必须要有表名

　　　　　　b. 必须要有对主键的描述
　　　　　　c. 给所有实例化的类一个表结构信息

　　　　5. 这里就需要一个元类来限制类中必须有上面这些内容   == 每一个表必须要有的内容也就有了

class ModlesMetaclas(type):
    def __new__(cls, name, bases, attrs):
        if name == ‘Modles‘:
            return type.__new__(cls, name, bases, attrs)  # 当类名为基类时就不要拦截
        table_name = attrs.get(‘table_name‘, None)
        if not table_name:
            table_name = name
        primary_key = None
        mappings = dict()
        for k, v in attrs.items():
            if isinstance(v, Filed):
                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‘] = table_name
        attrs[‘primary_key‘] = primary_key
        attrs[‘mappings‘] = mappings
        return type.__new__(cls, name, bases, attrs)


class Modles(dict, metaclass=ModlesMetaclas):
    def __init__(self, **kwargs):  # 实例对象时必须是关键字传参 给User实例对象时用的
        super().__init__(**kwargs)

    def __setattr__(self, key, value):  # 字典中有这个key就覆盖新的值，没有就装入到self这个字典中
        self[key] = value

    def __getattr__(self, item):
        try:
            return self[item]  # dic.name时【item类比是name】返回字典的value值
        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)

class User(Modles):

    table_name=‘user‘
    id=IntegerFileld(‘id‘,primary_key=True,default=0)
    name=StringFiled(‘name‘)
    password=StringFiled(‘password‘)

class Notic(Modles):
    table_name=‘notice‘
    id = IntegerFileld(‘id‘,primary_key=True)
    name=StringFiled(‘name‘)
    content=StringFiled(‘content‘)
    user_id=IntegerFileld(‘user_id‘)

　　　6. 另外实例化的User类【表】中的每一个属性【列】如id、name等都必须要有类型如int类型、char类型，那就来定义一些类吧：

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


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