Python+sqlite

如题，最近撸了一个小工具file-remark。考虑到数据规模和性能的关系使用json文件的方式存储显得不再合适，于是搜索了一下，尝试使用sqlite数据库来实现。

值得欣喜的是Python是自带sqlite3库的，这回省去一些使用第三方库的麻烦。

废话不多说，直接上代码：

import sqlite3
import os
from sqlite3.dbapi2 import Connection, Cursor
from typing import Any

from file_remark.config import Config
from file_remark.user_exception import UserException


class MyDB:
    '''数据库类（单例）'''
    DB_FILE_NAME = 'my_db.db'
    inited = False

    def __new__(cls) -> Any:
        if not hasattr(cls, "__instance"):
            setattr(cls, "__instance", super().__new__(cls))
        return getattr(cls, "__instance")

    def __init__(self) -> None:
        if not MyDB.inited:
            self.init_db()
            db_file = self.get_db_file()
            self.conn: Connection = sqlite3.connect(db_file)  # 数据库连接
            def dict_factory(cursor, row):
                d = {}
                for idx, col in enumerate(cursor.description):
                    d[col[0]] = row[idx]
                return d
            self.conn.row_factory = dict_factory
            # 将数据库中的配置数据写入config对象
            self.__load_config_from_db()
            MyDB.inited = True

    def get_db_file(self) -> str:
        '''获取数据库文件路径
        return 数据库文件路径
        '''
        return MyDB.get_db_file()

    @classmethod
    def get_db_file(cls):
        '''获取数据库文件路径
        return 数据库文件路径
        '''
        config: Config = Config()
        db_file = config.get_path_split().join(
            (config.get_data_dir(), MyDB.DB_FILE_NAME))
        return db_file

    def init_db(self) -> None:
        '''如果数据库文件不存在，初始化数据库文件'''
        db_file = self.get_db_file()
        if not os.path.exists(db_file):
            conn = sqlite3.connect(db_file)
            sql = self.__get_sql('files')
            conn.execute(sql)
            sql = self.__get_sql('config')
            conn.execute(sql)
            sql = self.__get_sql('config_data')
            conn.execute(sql)
            conn.commit()
            conn.close()

    def execute(self, sql: str) -> None:
        self.conn.execute(sql)
        self.conn.commit()

    def query(self, sql: str) -> list:
        cursor: Cursor = self.conn.cursor()
        cursor.execute(sql)
        result = cursor.fetchall()
        cursor.close()
        return result

    def query_one(self, sql: str) -> list:
        '''仅查询一条数据'''
        cursor: Cursor = self.conn.cursor()
        cursor.execute(sql)
        result = cursor.fetchone()
        cursor.close()
        return result

    def reset_db(self):
        '''初始化数据库'''
        self.conn.close()
        db_file = self.get_db_file()
        os.remove(db_file)
        self.inited=False
        self.__init__()

    def __del__(self):
        '''数据库对象注销时断开连接'''
        self.conn.close()
        super().__del__()

    def __get_sql(self, file_name: str) -> str:
        '''从sql文件获取sql
        file_name: sql文件名
        return sql内容
        '''
        config: Config = Config()
        sql_dir = config.get_sql_dir()
        sql_file = sql_dir+config.get_path_split()+file_name+'.sql'
        if not os.path.exists(sql_file):
            error_msg = "文件或目录{}不存在".format(sql_file)
            raise UserException(UserException.CODE_NO_PATH, error_msg)
        sql_content = ''
        with open(sql_file, 'r', encoding='UTF-8') as fopen:
            sql_content = fopen.read()
        return sql_content

    def __load_config_from_db(self) -> None:
        '''从数据库加载配置数据到config对象'''
        config: Config = Config()
        sql = 'SELECT * FROM config;'
        db_configs = self.query(sql)
        for db_config in db_configs:
            config.set_db_config(db_config['name'], db_config['value'])

这是我编写的一个用于支持数据库的module类。

特点有：

• 是单例，可以避免不必要的数据库连接浪费。
• 是长连接，在初始化方法中创建连接，在析构函数中释放连接。在频繁查询数据库时可以保持较好的性能。
• 封装了用于查询数据的query方法，以及用于修改数据的execute方法，简化SQL操作。
• 返回的数据为常见的键-值对形式，而单纯的值，方便后续的数据处理。

实际上个类实现除了必要的代码以外还因为项目的需要添加了其它的功能代码，想参考代码的同学可以自行删减，我这里偷懒就不做删减了。

想交流的话可以直接留言。