通過python封裝SQLite3的示例代碼
更新時間：2024年11月15日 09:21:40 作者：天寧
這篇文章主要介紹了使用 Python 封裝 SQLite3 的相關(guān)內(nèi)容,包括創(chuàng)建數(shù)據(jù)庫實例、創(chuàng)建表、各種插入（單個、批量、字典等）、刪除（按條件、按 ID 等）、修改（多種方式）、查詢（單條、多條、分頁等）的示例及對應(yīng)的實現(xiàn)代碼,需要的朋友可以參考下
創(chuàng)建數(shù)據(jù)庫實例
from main import SQLiteDB
?
db = SQLiteDB("example.db")
使用execute直接執(zhí)行SQL
db.execute("INSERT INTO user (name, amount, createtime) VALUES (?, ?, ?)",("張三", 25.6, '2023-07-01 15:25:30'))
創(chuàng)建表
# 創(chuàng)建表
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")
修改示例
# 基礎(chǔ)更新
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,))
?
# 批量更新字典數(shù)據(jù)
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)
聯(lián)表查詢示例
假設(shè)有兩個表：user和orders
內(nèi)連接查詢示例
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):
"""初始化數(shù)據(jù)庫連接"""
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"連接數(shù)據(jù)庫時出錯: {e}")
def execute(self, sql, params=None):
"""執(zhí)行SQL查詢
參數(shù):
sql: SQL語句
params: SQL參數(shù)，用于參數(shù)化查詢
返回:
執(zhí)行成功返回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"執(zhí)行查詢時出錯: {e}")
return False
# 創(chuàng)建表
def create_table(self, table_name, fields):
"""創(chuàng)建數(shù)據(jù)表
參數(shù):
table_name: 表名
fields: 字段定義列表，每個元素是一個元組 (字段名, 類型定義)
返回:
執(zhí)行成功返回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"創(chuàng)建表時出錯: {e}")
return False
# 插入##########################################
def insert(self, table_name, fields, values):
"""插入數(shù)據(jù)
參數(shù):
table_name: 表名
fields: 字段名列表，例如 ['name', 'age']
values: 值列表，例如 ('張三', 25)
返回:
執(zhí)行成功返回True，失敗返回False
"""
try:
# 構(gòu)建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"插入數(shù)據(jù)時出錯: {e}")
self.conn.rollback() # 發(fā)生錯誤時回滾
return False
# 批量插入1
def batch_insert(self, table_name, fields, values_list):
"""批量插入數(shù)據(jù)
參數(shù):
table_name: 表名
fields: 字段名列表，例如 ['name', 'age']
values_list: 值列表，每個元素是一個元組，例如 [('張三', 25), ('李四', 30)]
返回:
執(zhí)行成功返回True，失敗返回False
"""
try:
# 構(gòu)建SQL語句
placeholders = ','.join(['?' for _ in fields])
fields_str = ','.join(fields)
sql = f"INSERT INTO {table_name} ({fields_str}) VALUES ({placeholders})"
# 執(zhí)行批量插入
self.cursor.executemany(sql, values_list)
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"批量插入數(shù)據(jù)時出錯: {e}")
self.conn.rollback() # 發(fā)生錯誤時回滾
return False
# 批量插入2
def insert_many_dict(self, table_name, dict_list):
"""使用字典列表批量插入數(shù)據(jù)
參數(shù):
table_name: 表名
dict_list: 字典列表，每個字典代表一行數(shù)據(jù)，例如：
[{'name': '張三', 'age': 25}, {'name': '李四', 'age': 30}]
返回:
執(zhí)行成功返回True，失敗返回False
"""
if not dict_list:
return False
try:
# 從第一個字典獲取字段名
fields = list(dict_list[0].keys())
# 轉(zhuǎn)換字典列表為值列表
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"處理字典數(shù)據(jù)時出錯: {e}")
return False
############################################################
# 刪除###########################################
def delete(self, table_name, condition, params=None):
"""刪除數(shù)據(jù)
參數(shù):
table_name: 表名
condition: WHERE條件語句，例如 "age > ?" 或 "name = ?"
params: 條件參數(shù)，例如 (20,) 或 ('張三',)
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-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"刪除數(shù)據(jù)時出錯: {e}")
self.conn.rollback()
return -1
def delete_by_id(self, table_name, id_value, id_field='id'):
"""根據(jù)ID刪除數(shù)據(jù)
參數(shù):
table_name: 表名
id_value: ID值
id_field: ID字段名，默認(rèn)為'id'
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-1
"""
return self.delete(table_name, f"{id_field} = ?", (id_value,))
def delete_many(self, table_name, id_list, id_field='id'):
"""批量刪除數(shù)據(jù)
參數(shù):
table_name: 表名
id_list: ID列表
id_field: ID字段名，默認(rèn)為'id'
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-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"批量刪除數(shù)據(jù)時出錯: {e}")
self.conn.rollback()
return -1
def truncate_table(self, table_name):
"""清空表數(shù)據(jù)
參數(shù):
table_name: 表名
返回:
執(zhí)行成功返回True，失敗返回False
"""
try:
self.cursor.execute(f"DELETE FROM {table_name}")
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"清空表數(shù)據(jù)時出錯: {e}")
self.conn.rollback()
return False
############################################################
# 更新###########################################
def update(self, table_name, fields, values, condition, condition_params=None):
"""更新數(shù)據(jù)
參數(shù):
table_name: 表名
fields: 要更新的字段列表，例如 ['name', 'age']
values: 新的值列表，例如 ('張三', 25)
condition: WHERE條件語句，例如 "id = ?"
condition_params: 條件參數(shù)，例如 (1,)
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-1
"""
try:
# 構(gòu)建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"更新數(shù)據(jù)時出錯: {e}")
self.conn.rollback()
return -1
def update_by_id(self, table_name, fields, values, id_value, id_field='id'):
"""根據(jù)ID更新數(shù)據(jù)
參數(shù):
table_name: 表名
fields: 要更新的字段列表，例如 ['name', 'age']
values: 新的值列表，例如 ('張三', 25)
id_value: ID值
id_field: ID字段名，默認(rèn)為'id'
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-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):
"""使用字典更新數(shù)據(jù)
參數(shù):
table_name: 表名
update_dict: 要更新的字段和值的字典，例如 {'name': '張三', 'age': 25}
condition: WHERE條件語句，例如 "id = ?"
condition_params: 條件參數(shù)，例如 (1,)
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-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'):
"""使用字典列表批量更新數(shù)據(jù)
參數(shù):
table_name: 表名
dict_list: 字典列表，每個字典必須包含id_field字段，例如：
[{'id': 1, 'name': '張三', 'age': 25},
{'id': 2, 'name': '李四', 'age': 30}]
id_field: ID字段名，默認(rèn)為'id'
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-1
"""
if not dict_list:
return 0
try:
# 從第一個字典獲取所有字段名（排除ID字段）
fields = [f for f in dict_list[0].keys() if f != id_field]
# 轉(zhuǎn)換字典列表為值列表
values_list = []
for d in dict_list:
# 確保字典中包含ID字段
if id_field not in d:
raise ValueError(f"字典中缺少 {id_field} 字段")
# 構(gòu)建值元組：先添加要更新的字段值，最后添加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"批量更新字典數(shù)據(jù)時出錯: {e}")
return -1
def batch_update(self, table_name, fields, values_list, id_field='id'):
"""批量更新數(shù)據(jù)
參數(shù):
table_name: 表名
fields: 要更新的字段列表，例如 ['name', 'age']
values_list: 值列表，每個元素是一個元組，包含新值和ID，例如 [('張三', 25, 1), ('李四', 30, 2)]
id_field: ID字段名，默認(rèn)為'id'
返回:
執(zhí)行成功返回受影響的行數(shù)，失敗返回-1
"""
try:
# 構(gòu)建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"批量更新數(shù)據(jù)時出錯: {e}")
self.conn.rollback()
return -1
############################################################
# 查詢
def fetch_all(self, sql, params=None):
"""獲取所有查詢結(jié)果
參數(shù):
sql: SQL查詢語句
params: SQL參數(shù)，用于參數(shù)化查詢
返回:
查詢結(jié)果列表，失敗返回空列表
"""
try:
if params:
self.cursor.execute(sql, params)
else:
self.cursor.execute(sql)
return self.cursor.fetchall()
except sqlite3.Error as e:
print(f"獲取數(shù)據(jù)時出錯: {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):
"""獲取單條查詢結(jié)果
參數(shù):
sql: SQL查詢語句
params: SQL參數(shù)，用于參數(shù)化查詢
返回:
單條查詢結(jié)果，失敗返回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"獲取數(shù)據(jù)時出錯: {e}")
return None
############################################################
# 銷毀對象時關(guān)閉數(shù)據(jù)庫連接
def __del__(self):
try:
self.execute("VACUUM;")
"""關(guān)閉數(shù)據(jù)庫連接"""
if self.conn:
self.cursor.close()
self.conn.close()
except sqlite3.Error as e:
pass
到此這篇關(guān)于通過python封裝SQLite3的示例代碼的文章就介紹到這了,更多相關(guān)python封裝SQLite3內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Python中如何使用sqlite3操作SQLite數(shù)據(jù)庫詳解
- 使用Python連接SQLite數(shù)據(jù)庫的操作步驟
- Python數(shù)據(jù)庫編程之SQLite和MySQL的實踐指南
- Python的sqlite3模塊中常用函數(shù)
- Python中SQLite數(shù)據(jù)庫的使用
- Python數(shù)據(jù)庫sqlite3圖文實例詳解
- Python使用sqlite3第三方庫讀寫SQLite數(shù)據(jù)庫的方法步驟
- Python練習(xí)之操作SQLite數(shù)據(jù)庫
- python處理SQLite數(shù)據(jù)庫的方法
- SQLite5-使用Python來讀寫數(shù)據(jù)庫
- Pandas使用SQLite3實戰(zhàn)
相關(guān)文章
這篇文章主要介紹了Python函數(shù)進(jìn)階與文件操作詳情，文章為榮啊主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下2022-08-08
這篇文章主要為大家介紹了Python解析pcap文件示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-07-07
使用python3.0?對接美團(tuán)接口的實現(xiàn)示例
本文主要介紹了python3.0?對接美團(tuán)接口的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-05-05
平穩(wěn)性是時間序列分析與預(yù)測的核心概念,在平穩(wěn)條件下,時間序列的統(tǒng)計特性（如均值）在時間維度上保持不變,僅存在隨機(jī)波動,但是時間序列通常會經(jīng)歷結(jié)構(gòu)性斷裂或變化,本文給大家介紹了Python實現(xiàn)時間序列變化點檢測功能,需要的朋友可以參考下2024-09-09
Python使用collections模塊實現(xiàn)擴(kuò)展數(shù)據(jù)類
Python?標(biāo)準(zhǔn)庫提供了一個?collections?模塊，里面提供了很多的數(shù)據(jù)類，在工作中使用這些類能夠簡化我們的開發(fā)，本文就來看看collections是如何實現(xiàn)擴(kuò)展數(shù)據(jù)類的吧2023-06-06
淺談TensorFlow中讀取圖像數(shù)據(jù)的三種方式
這篇文章主要介紹了淺談TensorFlow中讀取圖像數(shù)據(jù)的三種方式,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-06-06
python爬取B站關(guān)注列表及數(shù)據(jù)庫的設(shè)計與操作
這篇文章主要為大家介紹了python爬取B站關(guān)注列表及數(shù)據(jù)庫的設(shè)計與操作，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
這篇文章主要介紹了python getpass實現(xiàn)密文實例詳解,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2019-09-09
本篇文章主要介紹了PyQt5 pyqt多線程操作入門，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
從運(yùn)行效率與開發(fā)效率比較Python和C++
今天小編就為大家分享一篇關(guān)于從運(yùn)行效率與開發(fā)效率比較Python和C++，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2018-12-12