通过python封装SQLite3的示例代码
更新时间：2024年11月15日 09:21:40 作者：天宁
这篇文章主要介绍了使用 Python 封装 SQLite3 的相关内容,包括创建数据库实例、创建表、各种插入（单个、批量、字典等）、删除（按条件、按 ID 等）、修改（多种方式）、查询（单条、多条、分页等）的示例及对应的实现代码,需要的朋友可以参考下
创建数据库实例
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
到此这篇关于通过python封装SQLite3的示例代码的文章就介绍到这了,更多相关python封装SQLite3内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
python创建一个最简单http webserver服务器的方法
这篇文章主要介绍了python创建一个最简单http webserver服务器的方法,实例分析了Python操作http创建服务器端的相关技巧,需要的朋友可以参考下2015-05-05
Python库coala代码分析和自动化检查改进工具使用探索
Python的coala工具是一个强大的代码分析和自动化改进工具,它可以帮助开发人员自动化地检查代码并提供改进建议,本文将详细介绍coala的功能和用法,并提供丰富的示例代码来帮助大家深入了解它2024-01-01
这篇文章主要为大家详细介绍了Python如何利用cv2实现动态绘制圆和矩形的功能，文中的示例代码讲解详细，具有一定的参考价值，需要的可以参考一下2023-03-03
这篇文章主要介绍了基于Python中的turtle绘画星星和星空，turtle 是 Python 中自带的绘图模块，下文章关于turtle绘画星星和星空的详细内容,需要的朋友可以参考一下，可以当作学习小练习2022-03-03
今天小编就为大家分享一篇Django中反向生成models.py的实例讲解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-05-05
这篇文章主要介绍了解决Pycharm 运行后没有输出的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2021-02-02
这篇文章主要给大家介绍了关于Python中np.where()用法的相关资料,np.where()是NumPy库中的一个函数,主要用于根据条件从数组中选择元素,文中给出了详细的代码示例,需要的朋友可以参考下2023-08-08
这篇文章主要介绍了python如何破解压缩包密码，帮助大家更好的利用python处理文件，感兴趣的朋友可以了解下2020-10-10
Python OpenCV 图像区域轮廓标记(框选各种小纸条)
这篇文章主要介绍了Python OpenCV 图像区域轮廓标记(框选各种小纸条)，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2021-03-03
这篇文章主要介绍了python实现二次元图片展示,用了API端口相关的知识实现,下面详细的文章内容需要的小伙伴可以参考一下2022-02-02
最新评论