创建数据库实例
from main import SQLiteDB db = SQLiteDB("example.db")
使用execute直接执行SQL
db.execute("INSERT INTO user (name, amount, createtime) VALUES (?, ?, ?)",("张三", 25.6, '2023-07-01 15:25:30'))
创建表
# 创建表 db.execute(""" CREATE TABLE users( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER, email TEXT UNIQUE, created_at TEXT DEFAULT (datetime('now', 'localtime')) ); """)
插入示例
# 单个插入 db.insert("user",['name', 'amount', 'createtime'],('李四', 25.6, '2023-07-01 15:25:30')) # 元组列表批量插入 fields = ['name', 'amount', 'createtime'] values = [ ('用户1', 22, '2024-11-12 12:13:11'), ('用户2', 23, '2024-11-12 12:13:11'), ('用户3', 24, '2024-11-12 12:13:11') ] db.batch_insert('user', fields, values) # 字典列表批量插入 users_dict = [ {'name': '小明', 'amount': 22, 'createtime': '2024-11-12 12:13:11'}, {'name': '小红', 'amount': 24, 'createtime': '2024-11-12 12:13:11'}, {'name': '小张', 'amount': 26, 'createtime': '2024-11-12 12:13:11'} ] db.insert_many_dict('user', users_dict)
删除示例
# 按条件删除 affected_rows = db.delete("user", "age > ?", (30,)) # 按ID删除 db.delete_by_id("user", 1) # 批量删除 id_list = [1, 2, 3, 4, 5] db.delete_many("user", id_list) # 清空表 db.truncate_table("user")
修改示例
# 基础更新 db.update('users', ['name', 'age'], ('张三', 25), 'id = ?', (1,)) # 通过ID更新 db.update_by_id('users', ['name', 'age'], ('张三', 25), 1) # 使用字典更新 db.update_dict('users', {'name': '张三', 'age': 25}, 'id = ?', (1,)) # 批量更新字典数据 dict_list = [ {'id': 1, 'name': '张三', 'age': 25, 'email': 'zhangsan@example.com'}, {'id': 2, 'name': '李四', 'age': 30, 'email': 'lisi@example.com'} ] db.batch_update_dict('users', dict_list) # 批量更新 values_list = [ ('张三', 25, 1), ('李四', 30, 2) ] db.batch_update('users', ['name', 'age'], values_list)
查询示例
# 查询单条记录 result = db.fetch_one("SELECT *FROM user WHERE name = ?", ("张三",)) # 分页查询 db.fetch_page("SELECT * FROM user", 3, 2) # 查询多条记录 results = db.fetch_all("SELECT *FROM user LIMIT 5") for row in results: print(row) # 条件查询 results = db.fetch_all("SELECT *FROM user WHERE amount > ?", (20,)) for row in results: print(row)
联表查询示例
假设有两个表:user和orders
内连接查询示例
sql = """ SELECT u.name, o.order_number, o.amount FROM user u INNER JOIN orders o ON u.id = o.user_id WHERE o.amount > ? """ results = db.fetch_all(sql, (100,)) for row in results: print(row)
左连接查询示例
sql = """ SELECT u.name, COUNT(o.id) as order_count FROM user u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.id """ results = db.fetch_all(sql) for row in results: print(row)
SQLiteDB
import sqlite3 class SQLiteDB: def __init__(self, db_name): """初始化数据库连接""" self.conn = None self.cursor = None try: self.conn = sqlite3.connect(db_name) self.cursor = self.conn.cursor() except sqlite3.Error as e: print(f"连接数据库时出错: {e}") def execute(self, sql, params=None): """执行SQL查询 参数: sql: SQL语句 params: SQL参数,用于参数化查询 返回: 执行成功返回True,失败返回False """ try: if params: self.cursor.execute(sql, params) else: self.cursor.execute(sql) self.conn.commit() return True except sqlite3.Error as e: print(f"执行查询时出错: {e}") return False # 创建表 def create_table(self, table_name, fields): """创建数据表 参数: table_name: 表名 fields: 字段定义列表,每个元素是一个元组 (字段名, 类型定义) 返回: 执行成功返回True,失败返回False """ try: fields_str = ', '.join([f"{name} {definition}" for name, definition in fields]) sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({fields_str})" return self.execute(sql) except sqlite3.Error as e: print(f"创建表时出错: {e}") return False # 插入########################################## def insert(self, table_name, fields, values): """插入数据 参数: table_name: 表名 fields: 字段名列表,例如 ['name', 'age'] values: 值列表,例如 ('张三', 25) 返回: 执行成功返回True,失败返回False """ try: # 构建SQL语句 placeholders = ','.join(['?' for _ in fields]) fields_str = ','.join(fields) sql = f"INSERT INTO {table_name} ({fields_str}) VALUES({placeholders})" return self.execute(sql, values) except sqlite3.Error as e: print(f"插入数据时出错: {e}") self.conn.rollback() # 发生错误时回滚 return False # 批量插入1 def batch_insert(self, table_name, fields, values_list): """批量插入数据 参数: table_name: 表名 fields: 字段名列表,例如 ['name', 'age'] values_list: 值列表,每个元素是一个元组,例如 [('张三', 25), ('李四', 30)] 返回: 执行成功返回True,失败返回False """ try: # 构建SQL语句 placeholders = ','.join(['?' for _ in fields]) fields_str = ','.join(fields) sql = f"INSERT INTO {table_name} ({fields_str}) VALUES ({placeholders})" # 执行批量插入 self.cursor.executemany(sql, values_list) self.conn.commit() return True except sqlite3.Error as e: print(f"批量插入数据时出错: {e}") self.conn.rollback() # 发生错误时回滚 return False # 批量插入2 def insert_many_dict(self, table_name, dict_list): """使用字典列表批量插入数据 参数: table_name: 表名 dict_list: 字典列表,每个字典代表一行数据,例如: [{'name': '张三', 'age': 25}, {'name': '李四', 'age': 30}] 返回: 执行成功返回True,失败返回False """ if not dict_list: return False try: # 从第一个字典获取字段名 fields = list(dict_list[0].keys()) # 转换字典列表为值列表 values_list = [tuple(d.values()) for d in dict_list] return self.batch_insert(table_name, fields, values_list) except Exception as e: print(f"处理字典数据时出错: {e}") return False ############################################################ # 删除########################################### def delete(self, table_name, condition, params=None): """删除数据 参数: table_name: 表名 condition: WHERE条件语句,例如 "age > ?" 或 "name = ?" params: 条件参数,例如 (20,) 或 ('张三',) 返回: 执行成功返回受影响的行数,失败返回-1 """ try: sql = f"DELETE FROM {table_name} WHERE {condition}" self.cursor.execute(sql, params or ()) self.conn.commit() return self.cursor.rowcount except sqlite3.Error as e: print(f"删除数据时出错: {e}") self.conn.rollback() return -1 def delete_by_id(self, table_name, id_value, id_field='id'): """根据ID删除数据 参数: table_name: 表名 id_value: ID值 id_field: ID字段名,默认为'id' 返回: 执行成功返回受影响的行数,失败返回-1 """ return self.delete(table_name, f"{id_field} = ?", (id_value,)) def delete_many(self, table_name, id_list, id_field='id'): """批量删除数据 参数: table_name: 表名 id_list: ID列表 id_field: ID字段名,默认为'id' 返回: 执行成功返回受影响的行数,失败返回-1 """ try: placeholders = ','.join(['?' for _ in id_list]) sql = f"DELETE FROM {table_name} WHERE {id_field} IN ({placeholders})" self.cursor.execute(sql, id_list) self.conn.commit() return self.cursor.rowcount except sqlite3.Error as e: print(f"批量删除数据时出错: {e}") self.conn.rollback() return -1 def truncate_table(self, table_name): """清空表数据 参数: table_name: 表名 返回: 执行成功返回True,失败返回False """ try: self.cursor.execute(f"DELETE FROM {table_name}") self.conn.commit() return True except sqlite3.Error as e: print(f"清空表数据时出错: {e}") self.conn.rollback() return False ############################################################ # 更新########################################### def update(self, table_name, fields, values, condition, condition_params=None): """更新数据 参数: table_name: 表名 fields: 要更新的字段列表,例如 ['name', 'age'] values: 新的值列表,例如 ('张三', 25) condition: WHERE条件语句,例如 "id = ?" condition_params: 条件参数,例如 (1,) 返回: 执行成功返回受影响的行数,失败返回-1 """ try: # 构建SET子句 set_clause = ','.join([f"{field} = ?" for field in fields]) sql = f"UPDATE {table_name} SET {set_clause} WHERE {condition}" # 合并values和condition_params params = list(values) if condition_params: params.extend(condition_params) self.cursor.execute(sql, params) self.conn.commit() return self.cursor.rowcount except sqlite3.Error as e: print(f"更新数据时出错: {e}") self.conn.rollback() return -1 def update_by_id(self, table_name, fields, values, id_value, id_field='id'): """根据ID更新数据 参数: table_name: 表名 fields: 要更新的字段列表,例如 ['name', 'age'] values: 新的值列表,例如 ('张三', 25) id_value: ID值 id_field: ID字段名,默认为'id' 返回: 执行成功返回受影响的行数,失败返回-1 """ return self.update(table_name, fields, values, f"{id_field} = ?", (id_value,)) def update_dict(self, table_name, update_dict, condition, condition_params=None): """使用字典更新数据 参数: table_name: 表名 update_dict: 要更新的字段和值的字典,例如 {'name': '张三', 'age': 25} condition: WHERE条件语句,例如 "id = ?" condition_params: 条件参数,例如 (1,) 返回: 执行成功返回受影响的行数,失败返回-1 """ fields = list(update_dict.keys()) values = list(update_dict.values()) return self.update(table_name, fields, values, condition, condition_params) def batch_update_dict(self, table_name, dict_list, id_field='id'): """使用字典列表批量更新数据 参数: table_name: 表名 dict_list: 字典列表,每个字典必须包含id_field字段,例如: [{'id': 1, 'name': '张三', 'age': 25}, {'id': 2, 'name': '李四', 'age': 30}] id_field: ID字段名,默认为'id' 返回: 执行成功返回受影响的行数,失败返回-1 """ if not dict_list: return 0 try: # 从第一个字典获取所有字段名(排除ID字段) fields = [f for f in dict_list[0].keys() if f != id_field] # 转换字典列表为值列表 values_list = [] for d in dict_list: # 确保字典中包含ID字段 if id_field not in d: raise ValueError(f"字典中缺少 {id_field} 字段") # 构建值元组:先添加要更新的字段值,最后添加ID值 values = tuple(d[f] for f in fields) values += (d[id_field],) values_list.append(values) return self.batch_update(table_name, fields, values_list, id_field) except Exception as e: print(f"批量更新字典数据时出错: {e}") return -1 def batch_update(self, table_name, fields, values_list, id_field='id'): """批量更新数据 参数: table_name: 表名 fields: 要更新的字段列表,例如 ['name', 'age'] values_list: 值列表,每个元素是一个元组,包含新值和ID,例如 [('张三', 25, 1), ('李四', 30, 2)] id_field: ID字段名,默认为'id' 返回: 执行成功返回受影响的行数,失败返回-1 """ try: # 构建SET子句 set_clause = ','.join([f"{field} = ?" for field in fields]) sql = f"UPDATE {table_name} SET {set_clause} WHERE {id_field} = ?" self.cursor.executemany(sql, values_list) self.conn.commit() return self.cursor.rowcount except sqlite3.Error as e: print(f"批量更新数据时出错: {e}") self.conn.rollback() return -1 ############################################################ # 查询 def fetch_all(self, sql, params=None): """获取所有查询结果 参数: sql: SQL查询语句 params: SQL参数,用于参数化查询 返回: 查询结果列表,失败返回空列表 """ try: if params: self.cursor.execute(sql, params) else: self.cursor.execute(sql) return self.cursor.fetchall() except sqlite3.Error as e: print(f"获取数据时出错: {e}") return [] # 分页查询 def fetch_page(self, sql, page_num, page_size, params=None): page_sql = f" limit {(page_num - 1) * page_size},{page_size}" print(sql + page_sql) return self.fetch_all(sql + page_sql, params) def fetch_one(self, sql, params=None): """获取单条查询结果 参数: sql: SQL查询语句 params: SQL参数,用于参数化查询 返回: 单条查询结果,失败返回None """ try: if params: self.cursor.execute(sql, params) else: self.cursor.execute(sql) return self.cursor.fetchone() except sqlite3.Error as e: print(f"获取数据时出错: {e}") return None ############################################################ # 销毁对象时关闭数据库连接 def __del__(self): try: self.execute("VACUUM;") """关闭数据库连接""" if self.conn: self.cursor.close() self.conn.close() except sqlite3.Error as e: pass
作者:天宁
链接:https://juejin.cn/post/7437029365526183973