简单使用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()