简单实现python sqlite 的增删改查封装

import sqlite3
import datetime
from typing import Optional, List, Dict

class SQLiteDB:
    def __init__(self, db_name: str = "data.db"):
        """初始化数据库连接"""
        self.db_name = db_name
        self.conn = None
        self.cursor = None
        self._connect()
        self._create_table()  # 确保表存在

    def _connect(self):
        """建立数据库连接"""
        try:
            self.conn = sqlite3.connect(self.db_name)
            self.cursor = self.conn.cursor()
            print(f"成功连接数据库：{self.db_name}")
        except sqlite3.Error as e:
            print(f"数据库连接失败：{e}")

    def _create_table(self):
        """创建数据表（id自增、text、created_at）"""
        create_sql = """
        CREATE TABLE IF NOT EXISTS content (
            id INTEGER PRIMARY KEY AUTOINCREMENT,  -- 自增主键
            text TEXT NOT NULL,  -- 内容字段（非空）
            created_at TIMESTAMP NOT NULL  -- 创建时间（非空）
        );
        """
        try:
            self.cursor.execute(create_sql)
            self.conn.commit()
            print("数据表创建成功（或已存在）")
        except sqlite3.Error as e:
            print(f"创建数据表失败：{e}")
            self.conn.rollback()

    def insert_data(self, text: str, created_at: Optional[datetime.datetime] = None) -> Optional[int]:
        """
        插入数据
        :param text: 内容字段
        :param created_at: 创建时间（默认当前时间）
        :return: 插入数据的自增ID，失败返回None
        """
        if not created_at:
            created_at = datetime.datetime.now()  # 默认使用当前时间
        # 格式化时间为 SQLite 支持的字符串格式（ISO 8601）
        created_at_str = created_at.strftime("%Y-%m-%d %H:%M:%S")
        
        insert_sql = """
        INSERT INTO content (text, created_at)
        VALUES (?, ?);
        """
        try:
            self.cursor.execute(insert_sql, (text, created_at_str))
            self.conn.commit()
            print(f"数据插入成功，自增ID：{self.cursor.lastrowid}")
            return self.cursor.lastrowid  # 返回自增的id
        except sqlite3.Error as e:
            print(f"插入数据失败：{e}")
            self.conn.rollback()
            return None

    def query_data(self, id: Optional[int] = None) -> List[Dict]:
        """
        查询数据
        :param id: 可选，根据ID查询单条数据；不填则查询所有
        :return: 结果列表（字典格式）
        """
        query_sql = "SELECT id, text, created_at FROM content"
        params = ()
        if id:
            query_sql += " WHERE id = ?"
            params = (id,)
        
        try:
            self.cursor.execute(query_sql, params)
            columns = [desc[0] for desc in self.cursor.description]  # 获取字段名
            results = [dict(zip(columns, row)) for row in self.cursor.fetchall()]
            print(f"查询到 {len(results)} 条数据")
            return results
        except sqlite3.Error as e:
            print(f"查询数据失败：{e}")
            return []

    def update_data(self, id: int, new_text: str) -> bool:
        """
        更新数据（根据ID修改text字段）
        :param id: 要更新的数据ID
        :param new_text: 新的内容
        :return: 成功返回True，失败返回False
        """
        update_sql = """
        UPDATE content
        SET text = ?
        WHERE id = ?;
        """
        try:
            self.cursor.execute(update_sql, (new_text, id))
            self.conn.commit()
            if self.cursor.rowcount > 0:
                print(f"ID为 {id} 的数据更新成功")
                return True
            else:
                print(f"未找到ID为 {id} 的数据")
                return False
        except sqlite3.Error as e:
            print(f"更新数据失败：{e}")
            self.conn.rollback()
            return False

    def delete_data(self, id: int) -> bool:
        """
        删除数据（根据ID）
        :param id: 要删除的数据ID
        :return: 成功返回True，失败返回False
        """
        delete_sql = "DELETE FROM content WHERE id = ?;"
        try:
            self.cursor.execute(delete_sql, (id,))
            self.conn.commit()
            if self.cursor.rowcount > 0:
                print(f"ID为 {id} 的数据删除成功")
                return True
            else:
                print(f"未找到ID为 {id} 的数据")
                return False
        except sqlite3.Error as e:
            print(f"删除数据失败：{e}")
            self.conn.rollback()
            return False

    def close(self):
        """关闭数据库连接"""
        if self.conn:
            self.conn.close()
            print("数据库连接已关闭")

# ------------------------------
# 使用示例
# ------------------------------
if __name__ == "__main__":
    # 1. 初始化数据库（自动创建data.db文件）
    db = SQLiteDB(db_name="content.db")

    # 2. 插入数据（默认使用当前时间）
    db.insert_data(text="第一条测试数据")
    db.insert_data(text="第二条带自定义时间的数据", created_at=datetime.datetime(2025, 11, 6, 10, 30, 0))

    # 3. 查询所有数据
    print("\n=== 查询所有数据 ===")
    all_data = db.query_data()
    for data in all_data:
        print(data)

    # 4. 根据ID查询单条数据（假设插入了ID=1的数据）
    print("\n=== 查询ID=1的数据 ===")
    single_data = db.query_data(id=1)
    print(single_data)

    # 5. 更新数据（修改ID=1的text字段）
    print("\n=== 更新ID=1的数据 ===")
    db.update_data(id=1, new_text="第一条测试数据（已更新）")

    # 6. 再次查询验证更新
    print("\n=== 更新后查询ID=1的数据 ===")
    print(db.query_data(id=1))

    # 7. 删除数据（删除ID=2的数据）
    print("\n=== 删除ID=2的数据 ===")
    db.delete_data(id=2)

    # 8. 最终查询所有数据
    print("\n=== 最终所有数据 ===")
    final_data = db.query_data()
    for data in final_data:
        print(data)

    # 9. 关闭连接
    db.close()

测试日志：

成功连接数据库：content.db
数据表创建成功（或已存在）
数据插入成功，自增ID：1
数据插入成功，自增ID：2

=== 查询所有数据 ===
查询到 2 条数据
{'id': 1, 'text': '第一条测试数据', 'created_at': '2025-11-06 09:57:19'}
{'id': 2, 'text': '第二条带自定义时间的数据', 'created_at': '2025-11-06 10:30:00'}

=== 查询ID=1的数据 ===
查询到 1 条数据
[{'id': 1, 'text': '第一条测试数据', 'created_at': '2025-11-06 09:57:19'}]

=== 更新ID=1的数据 ===
ID为 1 的数据更新成功

=== 更新后查询ID=1的数据 ===
查询到 1 条数据
[{'id': 1, 'text': '第一条测试数据（已更新）', 'created_at': '2025-11-06 09:57:19'}]

=== 删除ID=2的数据 ===
ID为 2 的数据删除成功

=== 最终所有数据 ===
查询到 1 条数据
{'id': 1, 'text': '第一条测试数据（已更新）', 'created_at': '2025-11-06 09:57:19'}
数据库连接已关闭