SQLite 返回字典数据

import os
import sqlite3
from typing import Iterator, Dict, Any, List
from app.config.settings import DB_PATH, DB_NAME

class DatabaseManager():
    def __init__(self):

        self.connection = None
        self._should_close = False

        self.db_path = DB_PATH
        os.makedirs(self.db_path, exist_ok=True)
        self.db_filename = os.path.join(self.db_path, DB_NAME)

    def __enter__(self):
        if not self.connection:
            self.connection = self.create_connection()
            self._should_close = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._should_close and self.connection:
            try:
                self.connection.close()
            except Exception as e:
                print(f"Error closing connection: {e}")

    def _database_exists(self):
        return os.path.exists(self.db_filename)

    def _create_database(self):
        try:
            con = sqlite3.connect(self.db_filename)
            con.close()
        except sqlite3.Error as e:
            print(f"Create database error: {e}")
            raise

    def create_connection(self):

        def dict_factory(cursor: sqlite3.Cursor, row: tuple) -> Dict[str, Any]:
            """ Convert a row into a dictionary """
            return {col[0]: val for col, val in zip(cursor.description, row)}

        if not self.connection:
            try:
                self.connection = sqlite3.connect(self.db_filename)
                self.connection.row_factory = dict_factory
                self.connection.execute("PRAGMA journal_mode = WAL")
            except sqlite3.Error as e:
                print(f"Connection error: {e}")
                raise
        return self.connection

    def fetch_query(self, query, params=None, single=False):
        result = None
        if self.connection:
            try:
                with self.connection:
                    cursor = self.connection.cursor()
                    cursor.execute(query, params or ())
                    if single:
                        result = cursor.fetchone()
                    else:
                        result = cursor.fetchall()

            except sqlite3.Error as e:
                print(f"Query execution error: {e}")
                return None
        else:
            print("No connection established.")
        return result

    def execute_query(self, query, params=None):
        if self.connection:
            try:
                with self.connection:
                    cursor = self.connection.cursor()
                    cursor.execute(query, params or ())
                    self.connection.commit()
                    return True
            except sqlite3.Error as e:
                print(f"Execution error: {e}")
                self.connection.rollback()
                return None
        else:
            print("No connection established.")
        return None

    def close_connection(self):
        if self.connection:
            self.connection.close()