功能描述

初始化时需传入数据库文件
提供执行单条sql语句的方法
提供执行sql文件的方法
执行单条sql或sql文件时，是自动打开数据库，执行后自动关闭数据库，若不需要该功能，可自行注释 self.__open_db()和self.__close_db()语句

代码实现

# _*_coding:utf-8_*_
'''
@File    :   sqlite.py
@Time    :   2023/05/02 18:01:41
@Author  :   Hu Guoqiang
@Version :   1.0
@Contact :   jayhgq@outlook.com
@Desc    :   对sqlite3数据库操作的封装
'''

# libs import here
import sqlite3
import os

class sqlite_operate(object):
    def __init__(self,db):
        '''
            初始化类，需要提供数据库文件路径，若不是数据库文件则报错
        '''
        if os.path.isfile(db) and db.endswith('.db'):
            self.dbname = db
        else:
            raise IOError("IOError：数据库路径错误，无法初始化，请修改数据库文件路径")
    
    def __open_db(self):
        '''
            打开数据库并创建操作游标
        '''
        try:
            self.con = sqlite3.connect(self.dbname)
            self.cur = self.con.cursor()
        except Exception as e:
            return(e)
    
    def __close_db(self):
        '''
            提交事物后关闭游标并关闭数据库
        '''
        try:
            self.con.commit()
            self.cur.close()
            self.con.close()
        except Exception as e:
            return(e)
    
    def execute_sql_file(self,sqlfile):
        '''
            执行sql文件，需要提供sql文件路径。可自动打开/关闭初始化的数据库
        '''
        if os.path.isfile(sqlfile):
            self.__open_db()
            try:
                all_sql = ''
                with open(sqlfile,'r', encoding='utf-8') as f:
                    all_sql = f.readlines()
                all_sql_join = ''.join(all_sql)
                all_sql = all_sql_join.replace('\n',' ').split(';')
                for sql in all_sql:
                    self.cur.execute(sql)
                self.__close_db()
                return('success')
            except Exception as e:
                return(e)
        else:
            raise IOError('sql文件不存在')

    def execute_sql_single(self,sql):
        '''
            执行单一sql语句，如果是查询语句则查询所有结果，并返回列表list，可自动打开/关闭初始化的数据库
        '''
        self.__open_db()
        if 'select' not in sql and 'SELECT' not in sql:
            try:
                self.cur.execute(sql)
                self.__close_db()
                return('success')
            except Exception as e:
                return(e)
        elif  'select' in sql or 'SELECT' in sql:
            try:
                self.cur.execute(sql)
                res = self.cur.fetchall()
                self.__close_db()
                return(res)
            except Exception as e:
                return(e)
        else:
            raise IOError('执行语句不能为空')