python连接sqlite3工具类

本文介绍: 简单使用python链接sqlite3工具类，代码可根据场景自行抽象。

简单使用python连接sqlite3工具类，代码可根据场景自行抽象

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sqlite3


class SQLiteHandler:
    # 示例用法
    default_db_path = "example.db"

    def __init__(self, db_file=None):
        self.db_file = db_file or self.default_db_path
        self.conn = self.create_connection()
        self.initialize_database()

    def create_connection(self):
        # 创建数据库连接
        try:
            conn = sqlite3.connect(self.db_file)
            print(f"Connected to {self.db_file}")
            return conn
        except sqlite3.Error as e:
            print(e)
            return None

    def initialize_database(self):
        # 初始化数据库，包括创建表格
        if self.conn:
            create_table_sql = """
            CREATE TABLE IF NOT EXISTS example_table (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                age INTEGER
            );
            """
            self.create_table(create_table_sql)
        else:
            print("Failed to initialize database. Connection not established.")

    def create_table(self, create_table_sql):
        # 创建表格
        try:
            cursor = self.conn.cursor()
            cursor.execute(create_table_sql)
            print("Table created successfully")
        except sqlite3.Error as e:
            print(e)

    def execute_query(self, query, data=None):
        # 执行查询操作
        try:
            cursor = self.conn.cursor()
            if data:
                cursor.execute(query, data)
            else:
                cursor.execute(query)
            self.conn.commit()
            print("Query executed successfully")
        except sqlite3.Error as e:
            print(e)

    def fetch_all(self, query):
        # 获取所有记录
        try:
            cursor = self.conn.cursor()
            cursor.execute(query)
            rows = cursor.fetchall()
            return rows
        except sqlite3.Error as e:
            print(e)

    def delete_record(self, table_name, record_id):
        # 删除特定记录
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"DELETE FROM {table_name} WHERE id = ?;", (record_id,))
            self.conn.commit()
            print(f"Record with ID {record_id} deleted successfully from {table_name}")
        except sqlite3.Error as e:
            print(e)

    def delete_all_records(self, table_name):
        # 删除表格中的所有记录
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"DELETE FROM {table_name};")
            self.conn.commit()
            print(f"All records deleted successfully from {table_name}")
        except sqlite3.Error as e:
            print(e)

    def drop_table(self, table_name):
        # 删除表格
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"DROP TABLE IF EXISTS {table_name};")
            self.conn.commit()
            print(f"Table {table_name} dropped successfully")
        except sqlite3.Error as e:
            print(e)


# # 示例用法
# db_path = "example.db"
#
# # 实例化 SQLiteHandler
db_handler = SQLiteHandler()

# 插入数据
insert_query = "INSERT INTO example_table (name, age) VALUES (?, ?);"
data_to_insert = ("John Doe", 35)
db_handler.execute_query(insert_query, data_to_insert)

# 查询数据
select_query = "SELECT * FROM example_table;"
result = db_handler.fetch_all(select_query)
print(result)

# 更新数据
# update_query = "UPDATE example_table SET age=? WHERE name=?;"
# data_to_update = (31, "John Doe")
# db_handler.execute_query(update_query, data_to_update)

# 查询更新后的数据
result_after_update = db_handler.fetch_all(select_query)
print(result_after_update)

# 删除名为example_table的表格中ID为1的记录
# db_handler.delete_record("example_table", 3)

# 删除表格
# table_to_drop = "example_table"
# db_handler.drop_table(table_to_drop)

# 关闭数据库连接
db_handler.conn.close()