通過(guò)python封裝SQLite3的示例代碼

更新時(shí)間：2024年11月15日 09:21:40 作者：天寧

這篇文章主要介紹了使用 Python 封裝 SQLite3 的相關(guān)內(nèi)容,包括創(chuàng)建數(shù)據(jù)庫(kù)實(shí)例、創(chuàng)建表、各種插入（單個(gè)、批量、字典等）、刪除（按條件、按 ID 等）、修改（多種方式）、查詢（單條、多條、分頁(yè)等）的示例及對(duì)應(yīng)的實(shí)現(xiàn)代碼,需要的朋友可以參考下

創(chuàng)建數(shù)據(jù)庫(kù)實(shí)例

 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'))
     );
     """)

插入示例

 # 單個(gè)插入
 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,))
 ?
 # 通過(guò)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 = ?", ("張三",))
 
 # 分頁(yè)查詢
 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è)有兩個(gè)表：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ù)庫(kù)連接"""
        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ù)庫(kù)時(shí)出錯(cuò): {e}")

    def execute(self, sql, params=None):
        """執(zhí)行SQL查詢
        參數(shù):
            sql: SQL語(yǔ)句
            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í)行查詢時(shí)出錯(cuò): {e}")
            return False

    # 創(chuàng)建表
    def create_table(self, table_name, fields):
        """創(chuàng)建數(shù)據(jù)表
        參數(shù):
            table_name: 表名
            fields: 字段定義列表，每個(gè)元素是一個(gè)元組 (字段名, 類型定義)
        返回:
            執(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)建表時(shí)出錯(cuò): {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語(yǔ)句
            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ù)時(shí)出錯(cuò): {e}")
            self.conn.rollback()  # 發(fā)生錯(cuò)誤時(shí)回滾
            return False

    # 批量插入1
    def batch_insert(self, table_name, fields, values_list):
        """批量插入數(shù)據(jù)
        參數(shù):
            table_name: 表名
            fields: 字段名列表，例如 ['name', 'age']
            values_list: 值列表，每個(gè)元素是一個(gè)元組，例如 [('張三', 25), ('李四', 30)]
        返回:
            執(zhí)行成功返回True，失敗返回False
        """
        try:
            # 構(gòu)建SQL語(yǔ)句
            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ù)時(shí)出錯(cuò): {e}")
            self.conn.rollback()  # 發(fā)生錯(cuò)誤時(shí)回滾
            return False

    # 批量插入2
    def insert_many_dict(self, table_name, dict_list):
        """使用字典列表批量插入數(shù)據(jù)
        參數(shù):
            table_name: 表名
            dict_list: 字典列表，每個(gè)字典代表一行數(shù)據(jù)，例如：
                     [{'name': '張三', 'age': 25}, {'name': '李四', 'age': 30}]
        返回:
            執(zhí)行成功返回True，失敗返回False
        """
        if not dict_list:
            return False
        try:
            # 從第一個(gè)字典獲取字段名
            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ù)時(shí)出錯(cuò): {e}")
            return False

    ############################################################

    # 刪除###########################################
    def delete(self, table_name, condition, params=None):
        """刪除數(shù)據(jù)
        參數(shù):
            table_name: 表名
            condition: WHERE條件語(yǔ)句，例如 "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ù)時(shí)出錯(cuò): {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ù)時(shí)出錯(cuò): {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ù)時(shí)出錯(cuò): {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條件語(yǔ)句，例如 "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ù)時(shí)出錯(cuò): {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條件語(yǔ)句，例如 "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: 字典列表，每個(gè)字典必須包含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:
            # 從第一個(gè)字典獲取所有字段名（排除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ù)時(shí)出錯(cuò): {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: 值列表，每個(gè)元素是一個(gè)元組，包含新值和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ù)時(shí)出錯(cuò): {e}")
            self.conn.rollback()
            return -1

    ############################################################
    # 查詢
    def fetch_all(self, sql, params=None):
        """獲取所有查詢結(jié)果
        參數(shù):
            sql: SQL查詢語(yǔ)句
            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ù)時(shí)出錯(cuò): {e}")
            return []

    # 分頁(yè)查詢
    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查詢語(yǔ)句
            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ù)時(shí)出錯(cuò): {e}")
            return None

    ############################################################

    # 銷毀對(duì)象時(shí)關(guān)閉數(shù)據(jù)庫(kù)連接
    def __del__(self):
        try:
            self.execute("VACUUM;")
            """關(guān)閉數(shù)據(jù)庫(kù)連接"""
            if self.conn:
                self.cursor.close()
                self.conn.close()
        except sqlite3.Error as e:
            pass

到此這篇關(guān)于通過(guò)python封裝SQLite3的示例代碼的文章就介紹到這了,更多相關(guān)python封裝SQLite3內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: