sqlalchemy对sql的简单操作

﹏ヽ暗。殇╰゛Y 2022-06-17 08:44 280阅读 0赞

# -*-coding:utf-8 -*-
    __author__ = "ZJL"
    
    from sqlalchemy.pool import QueuePool
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker, scoped_session
    import traceback
    
    
    # 用于需要回滚和提交的操作
    def alter(func):
        def wrapper(self, *args, **kwargs):
            try:
                a = func(self, *args, **kwargs)
            except Exception as e:
                # 如果函数出现错误，不管执行多少数据都全部回滚
                self.session.rollback()
                print(traceback.format_exc())
                print(str(e))
                return traceback.format_exc()
            else:
                # 只有整个函数执行结束没有错误才会提交数据
                self.session.commit()
                return a
            finally:
                self.session.close()
    
        return wrapper
    
    
    # 用于不需要回滚和提交的操作
    def find(func):
        def wrapper(self, *args, **kwargs):
            try:
                return func(self, *args, **kwargs)
            except Exception as e:
                print(traceback.format_exc())
                print(str(e))
                return traceback.format_exc()
            finally:
                self.session.close()
    
        return wrapper
    
    
    # 通用操作
    def common(func):
        def wrapper(self, *args, **kwargs):
            try:
                a = func(self, *args, **kwargs)
            except Exception as e:
                # 发生错误会回滚和关闭
                self.session.rollback()
                self.session.close()
                print(traceback.format_exc())
                print(str(e))
                return traceback.format_exc()
            else:
                self.session.commit()
                return a
    
        return wrapper
    
    
    # 通用操作不带提交
    def common_no(func):
        def wrapper(self, *args, **kwargs):
            try:
                a = func(self, *args, **kwargs)
            except Exception as e:
                # 发生错误会回滚和关闭
                self.session.rollback()
                self.session.close()
                print(traceback.format_exc())
                print(str(e))
                return traceback.format_exc()
            else:
                return a
    
        return wrapper
    
    
    class MysqlManager(object):
        def __init__(self):
            self.engine = create_engine('mysql+pymysql://root:root123456@localhost:3306/dbtest', poolclass=QueuePool,
                                        pool_recycle=3600)
            # self.DB_Session = sessionmaker(bind=self.engine)
            # self.session = self.DB_Session()
            self.DB_Session = sessionmaker(bind=self.engine, autocommit=False, autoflush=True, expire_on_commit=False)
            self.db = scoped_session(self.DB_Session)
            self.session = self.db()
    
        @find
        def select_one(self, sql):
            a = self.session.execute(sql)
            return a.first()
    
        @find
        def select_one_dict(self, sql, keys):
            a = self.session.execute(sql)
            a = a.first()
            if len(keys) == len(a):
                data_dict = {}
                for k, v in zip(keys, a):
                    data_dict[k] = v
                return data_dict
            else:
                return False
    
        @find
        def select_one_list(self, sqls):
            if isinstance(sqls, list):
                data_list = []
                for sql in sqls:
                    a = self.session.execute(sql)
                    data_list.append(a.first())
                return data_list
            else:
                return False
    
        @find
        def select_one_dict_list(self, sqls, keys):
            if isinstance(sqls, list) and isinstance(keys, list):
                data_list = []
                for sql in sqls:
                    a = self.session.execute(sql)
                    data_list.append(a.first())
                print(data_list)
                print(keys)
                data_list_len = len(data_list)
                keys_len = len(keys)
                if data_list_len == keys_len:
                    sum_data = []
                    for k, d in zip(keys, data_list):
                        if not k:
                            k = ""
                        if not d:
                            d = ""
                        data_dict = {}
                        for x, y in zip(k, d):
                            data_dict[x] = y
                        sum_data.append(data_dict)
                    print(sum_data)
                    return sum_data
                else:
                    return False
            else:
                return False
    
        @find
        def select_all(self, sql):
            a = self.session.execute(sql)
            return a.fetchall()
    
        @find
        def select_all_dict(self, sql, keys):
            a = self.session.execute(sql)
            a = a.fetchall()
            lists = []
            for i in a:
                if len(keys) == len(i):
                    data_dict = {}
                    for k, v in zip(keys, i):
                        data_dict[k] = v
                    lists.append(data_dict)
                else:
                    return False
            return lists
    
        @find
        def count_one(self, sql):
            a = self.session.execute(sql)
            a = tuple(a.first())
            if isinstance(a, tuple) and a:
                return a[0]
            else:
                return False
    
        @alter
        def add_update_del(self, sql):
            self.session.execute(sql)
            return True
    
        @alter
        def add_update_del_list(self, sqls):
            if isinstance(sqls, list):
                for sql in sqls:
                    self.session.execute(sql)
                return True
            else:
                print("no list")
                return False
    
                # 用于频繁操作，这样就不会频繁连接和断开数据库，用common_sql需要手动关闭
    
        # 做查询以外操作必需code参数而且不能是find，比如更新可以是code="updata"
        @common
        def common_sql(self, sql, code="find"):
            if code != "find" and code != "findall":
                self.session.execute(sql)
                return True
            elif code == "findall":
                a = self.session.execute(sql)
                return a.fetchall()
            else:
                a = self.session.execute(sql)
                return a.first()
    
                # 用于频繁操作，这样就不会频繁连接和断开数据库，用common_no_sql需要手动提交和手动关闭
    
        # 做查询以外操作必需code参数而且不能是find，比如更新可以是code="updata"
        @common_no
        def common_no_sql(self, sql, code="find"):
            if code != "find" and code != "findall":
                self.session.execute(sql)
                return True
            elif code == "findall":
                a = self.session.execute(sql)
                return a.fetchall()
            else:
                a = self.session.execute(sql)
                return a.first()
    
                # 最原始的执行，应对复杂业务逻辑，所有操作手动执行
    
        def primary_sql(self, sql):
            return self.session.execute(sql)
    
            # 回滚
    
        def rollback(self):
            self.session.rollback()
    
            # 提交
    
        def commit(self):
            self.session.commit()
    
            # 关闭
    
        def close(self):
            self.session.close()
    
    
    ''''' 
    #以下例子 
    
    def main(): 
        #表字段，顺序不能错 
        dbkeys = ("id", "name") 
        mym = MysqlManager() 
        #统计 
        print(mym.count_one("select count(*)from dbtest.test;")) 
        #查询一条数据，返回字典结构： 
        print(mym.select_one_dict("SELECT %s FROM %s WHERE %s;"%("*","dbtest.test","id=3"), dbkeys)) 
        #查询多条数据，返回一个字典列表 
        print(mym.select_all_dict("SELECT %s FROM %s;"%("*","dbtest.test"), dbkeys)) 
        #查询单条数据 
        print(mym.select_one("SELECT %s FROM %s WHERE %s;"%("*","dbtest.test","id=3"))) 
        #查询多条数据 
        print(mym.select_all("SELECT %s FROM %s;"%("*","dbtest.test"))) 
        #插入数据，成功返回True，失败返回False或者报错信息 
        sql_ins='INSERT INTO %s VALUES (\'%s\',\'%s\');'%("dbtest.test","654321","zzzzxxx") 
        # print(mym.add_update_del(sql_ins)) 
        #删除数据，成功返回True，失败返回False或者报错信息 
        name="zzzz" 
        sql_del = "DELETE FROM %s WHERE name=\'%s\';"%("dbtest.test",name) 
        # print(mym.add_update_del(sql_del)) 
        #修改数据，成功返回True，失败返回False或者报错信息 
        sql_up = "UPDATE %s SET id=\'%s\' WHERE id=\'%s\'"%("dbtest.test","123456","4") 
        # print(mym.add_update_del(sql_up)) 
    
    
        #执行多条sql使用add_update_del_list 
        sql_1 = 'INSERT INTO %s VALUES (\'%s\',\'%s\');' % ("dbtest.test", "90", "09zzz") 
        sql_2 = 'INSERT INTO %s VALUES (\'%s\',\'%s\');' % ("dbtest.test", "91", "19zzz") 
        sql_3 = 'INSERT INTO %s VALUES (\'%s\',\'%s\');' % ("dbtest.test", "92", "29zzz") 
        # print(mym.add_update_del_list([sql_1,sql_2,sql_3])) 
    
        #多条语句查询，返回一个list 
        print(mym.select_one_dict_list([text("SELECT * FROM `user` WHERE id=:id").params(id=userid),text("SELECT * FROM `user` WHERE id=:id").params(id=userid)])) 
    
        #多条语句查询，返回list，内容是字典 
        USER_TABLE=("id", "name","class") 
        print(mym.select_one_dict_list([text("SELECT * FROM `user` WHERE id=:id").params(id=userid),text("SELECT * FROM `user` WHERE id=:id").params(id=userid)],[USER_TABLE,USER_TABLE])) 
          #查询数据以后，可以直接通过对象取字段的值 
          xx = mm.select_one(text("SELECT * FROM `user` WHERE id=:id;").params(id=id)) 
          #取出phone字段的值 
          print(xx.phone) 
          #最原始的操作 
          mm.session.execute() 
    
    
    if __name__ == '__main__': 
        main() 
    '''