通过python封装SQLite3的示例代码！

admin • • Python学习

通过python封装SQLite3的示例代码！

这篇文章主要介绍了使用 Python 封装 SQLite3 的相关内容,包括创建数据库实例、创建表、各种插入（单个、批量、字典等）、删除（按条件、按 ID 等）、修改（多种方式）、查询（单条、多条、分页等）的示例及对应的实现代码,需要的朋友可以参考下。

创建数据库实例

1
2
3
from main import SQLiteDB
​
db = SQLiteDB("example.db")

 

使用execute直接执行SQL

1
db.execute("INSERT INTO user (name, amount, createtime) VALUES (?, ?, ?)",("张三", 25.6, '2023-07-01 15:25:30'))

创建表

1
2
3
4
5
6
7
8
9
10
# 创建表
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'))
);
""")

插入示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 单个插入
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)

删除示例

1
2
3
4
5
6
7
8
9
10
11
12
# 按条件删除
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")

修改示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 基础更新
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)

查询示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 查询单条记录
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

内连接查询示例

1
2
3
4
5
6
7
8
9
10
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)

左连接查询示例

1
2
3
4
5
6
7
8
9
10
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
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的示例代码的文章就介绍到这了。

 

 

学习资料见知识星球。

以上就是今天要分享的技巧，你学会了吗？若有什么问题，欢迎在下方留言。

快来试试吧，小琥 my21ke007。获取 1000个免费 Excel模板福利​​​​！

更多技巧， www.excelbook.cn

欢迎 加入 零售创新 知识星球，知识星球主要以数据分析、报告分享、数据工具讨论为主；

Excelbook.cn Excel技巧 SQL技巧 Python 学习！

你将获得：

1、价值上万元的专业的PPT报告模板。

2、专业案例分析和解读笔记。

3、实用的Excel、Word、PPT技巧。

4、VIP讨论群，共享资源。

5、优惠的会员商品。

6、一次付费只需99元，即可下载本站文章涉及的文件和软件。

文章版权声明 1、本网站名称：Excelbook
2、本站永久网址：http://www.excelbook.cn
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长王小琥进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报。
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END
分享
二维码
< <上一篇
下一篇>>