醋醋百科网

课程目标

• 理解数据库的基本概念和 SQLite 的特点。
• 掌握 Python 的 sqlite3 模块，进行基本的增删改查（CRUD）操作。
• 能够将 SQLite 数据库集成到实际项目中（如财务管理系统）。
• 培养数据库设计和 SQL 查询的初步能力。

课程时长

• 3 小时（1 小时理论讲解 + 1 小时代码实践 + 1 小时练习与答疑）

课程大纲

1. 理论讲解：数据库与 SQLite 基础（60 分钟）

1. 什么是数据库？
2. 数据库的定义：结构化存储数据的工具。
3. 关系型数据库 vs 非关系型数据库。
4. SQLite 简介：轻量级、嵌入式、无服务器的关系型数据库。
5. SQLite 的适用场景：小型项目、移动应用、本地开发。
6. SQLite 的核心概念
7. 表（Table）：存储数据的结构，包含行和列。
8. 主键（Primary Key）：唯一标识每条记录的字段。
9. SQL 语句：用于操作数据库的语言（CREATE、INSERT、SELECT、UPDATE、DELETE）。
10. sqlite3 模块：Python 内置的 SQLite 操作库。
11. Python 与 SQLite 的连接
12. 连接数据库：sqlite3.connect()。
13. 执行 SQL 语句：cursor.execute()。
14. 事务管理：commit() 和 rollback()。
15. 关闭连接：close()。
16. 安全性提示
17. 避免 SQL 注入：使用参数化查询（? 或 :name）。

2. 代码实践：掌握增删改查（60 分钟）

通过一个简单的示例，逐步演示如何使用 sqlite3 模块进行 CRUD 操作。

示例项目：个人财务记录数据库

• 需求：创建一个数据库，记录用户的收入和支出（包含金额、类别、日期、描述）。
• 数据库结构：
• sql
• CREATE TABLE transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, amount FLOAT NOT NULL, category TEXT NOT NULL, date TEXT NOT NULL, description TEXT );

代码示例：

python

import sqlite3
from datetime import datetime

# 1. 连接数据库（如果不存在则创建）
conn = sqlite3.connect("finance.db")
cursor = conn.cursor()

# 2. 创建表
cursor.execute("""
    CREATE TABLE IF NOT EXISTS transactions (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        amount FLOAT NOT NULL,
        category TEXT NOT NULL,
        date TEXT NOT NULL,
        description TEXT
    )
""")

# 3. 插入数据（CREATE）
def add_transaction(amount, category, date, description):
    cursor.execute("""
        INSERT INTO transactions (amount, category, date, description)
        VALUES (?, ?, ?, ?)
    """, (amount, category, date, description))
    conn.commit()

# 4. 查询数据（READ）
def get_all_transactions():
    cursor.execute("SELECT * FROM transactions")
    return cursor.fetchall()

def get_transactions_by_category(category):
    cursor.execute("SELECT * FROM transactions WHERE category = ?", (category,))
    return cursor.fetchall()

# 5. 更新数据（UPDATE）
def update_transaction(id, amount, category, date, description):
    cursor.execute("""
        UPDATE transactions
        SET amount = ?, category = ?, date = ?, description = ?
        WHERE id = ?
    """, (amount, category, date, description, id))
    conn.commit()

# 6. 删除数据（DELETE）
def delete_transaction(id):
    cursor.execute("DELETE FROM transactions WHERE id = ?", (id,))
    conn.commit()

# 示例操作
try:
    # 添加记录
    add_transaction(100.50, "Income", datetime.now().strftime("%Y-%m-%d"), "Salary")
    add_transaction(-20.00, "Expense", datetime.now().strftime("%Y-%m-%d"), "Coffee")

    # 查询所有记录
    print("All transactions:")
    for row in get_all_transactions():
        print(row)

    # 查询特定类别
    print("\nIncome transactions:")
    for row in get_transactions_by_category("Income"):
        print(row)

    # 更新记录
    update_transaction(1, 150.75, "Income", datetime.now().strftime("%Y-%m-%d"), "Bonus")

    # 删除记录
    delete_transaction(2)

    # 再次查询所有记录
    print("\nUpdated transactions:")
    for row in get_all_transactions():
        print(row)

except sqlite3.Error as e:
    print(f"Database error: {e}")
finally:
    # 关闭连接
    conn.close()

代码讲解：

• 数据库连接和表创建。
• 参数化查询防止 SQL 注入。
• CRUD 操作的实现。
• 异常处理和资源管理。

3. 练习：为财务管理系统添加 SQLite 支持（60 分钟）

练习目标： 将 SQLite 数据库集成到一个简单的财务管理系统，扩展之前的 CRUD 操作，支持以下功能：

1. 添加收入/支出记录。
2. 按类别或日期范围查询记录。
3. 修改已有记录。
4. 删除记录。
5. 统计某段时间内的总收入、总支出和净余额。

练习要求：

• 数据库名称：finance_manager.db。
• 表结构：
• sql
• CREATE TABLE transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT NOT NULL, -- 'Income' 或 'Expense' amount FLOAT NOT NULL, category TEXT NOT NULL, date TEXT NOT NULL, -- 格式为 'YYYY-MM-DD' description TEXT );
• 实现以下功能（提供函数接口）：
• add_transaction(type, amount, category, date, description)：添加记录。
• query_transactions(start_date, end_date)：查询指定日期范围的记录。
• update_transaction(id, type, amount, category, date, description)：更新记录。
• delete_transaction(id)：删除记录。
• get_summary(start_date, end_date)：返回总收入、总支出和净余额。
• 使用异常处理确保程序健壮性。
• 提供简单的命令行界面，允许用户交互式操作（可选）。

参考代码框架：

python

import sqlite3
from datetime import datetime

class FinanceManager:
    def __init__(self, db_name="finance_manager.db"):
        self.conn = sqlite3.connect(db_name)
        self.cursor = self.conn.cursor()
        self.create_table()

    def create_table(self):
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS transactions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                type TEXT NOT NULL,
                amount FLOAT NOT NULL,
                category TEXT NOT NULL,
                date TEXT NOT NULL,
                description TEXT
            )
        """)
        self.conn.commit()

    def add_transaction(self, type, amount, category, date, description):
        try:
            self.cursor.execute("""
                INSERT INTO transactions (type, amount, category, date, description)
                VALUES (?, ?, ?, ?, ?)
            """, (type, amount, category, date, description))
            self.conn.commit()
        except sqlite3.Error as e:
            print(f"Error adding transaction: {e}")

    def query_transactions(self, start_date, end_date):
        try:
            self.cursor.execute("""
                SELECT * FROM transactions
                WHERE date BETWEEN ? AND ?
            """, (start_date, end_date))
            return self.cursor.fetchall()
        except sqlite3.Error as e:
            print(f"Error querying transactions: {e}")
            return []

    def update_transaction(self, id, type, amount, category, date, description):
        try:
            self.cursor.execute("""
                UPDATE transactions
                SET type = ?, amount = ?, category = ?, date = ?, description = ?
                WHERE id = ?
            """, (type, amount, category, date, description, id))
            self.conn.commit()
        except sqlite3.Error as e:
            print(f"Error updating transaction: {e}")

    def delete_transaction(self, id):
        try:
            self.cursor.execute("DELETE FROM transactions WHERE id = ?", (id,))
            self.conn.commit()
        except sqlite3.Error as e:
            print(f"Error deleting transaction: {e}")

    def get_summary(self, start_date, end_date):
        try:
            self.cursor.execute("""
                SELECT type, SUM(amount) FROM transactions
                WHERE date BETWEEN ? AND ?
                GROUP BY type
            """, (start_date, end_date))
            summary = {"Income": 0.0, "Expense": 0.0}
            for row in self.cursor.fetchall():
                summary[row[0]] = row[1]
            balance = summary["Income"] - summary["Expense"]
            return summary["Income"], summary["Expense"], balance
        except sqlite3.Error as e:
            print(f"Error calculating summary: {e}")
            return 0.0, 0.0, 0.0

    def close(self):
        self.conn.close()

# 示例命令行界面
def main():
    fm = FinanceManager()
    while True:
        print("\n1. Add Transaction")
        print("2. Query Transactions")
        print("3. Update Transaction")
        print("4. Delete Transaction")
        print("5. Get Summary")
        print("6. Exit")
        choice = input("Choose an option: ")

        if choice == "1":
            type = input("Type (Income/Expense): ")
            amount = float(input("Amount: "))
            category = input("Category: ")
            date = input("Date (YYYY-MM-DD): ")
            description = input("Description: ")
            fm.add_transaction(type, amount, category, date, description)

        elif choice == "2":
            start_date = input("Start date (YYYY-MM-DD): ")
            end_date = input("End date (YYYY-MM-DD): ")
            transactions = fm.query_transactions(start_date, end_date)
            for t in transactions:
                print(t)

        elif choice == "3":
            id = int(input("Transaction ID: "))
            type = input("Type (Income/Expense): ")
            amount = float(input("Amount: "))
            category = input("Category: ")
            date = input("Date (YYYY-MM-DD): ")
            description = input("Description: ")
            fm.update_transaction(id, type, amount, category, date, description)

        elif choice == "4":
            id = int(input("Transaction ID: "))
            fm.delete_transaction(id)

        elif choice == "5":
            start_date = input("Start date (YYYY-MM-DD): ")
            end_date = input("End date (YYYY-MM-DD): ")
            income, expense, balance = fm.get_summary(start_date, end_date)
            print(f"Total Income: {income}")
            print(f"Total Expense: {expense}")
            print(f"Balance: {balance}")

        elif choice == "6":
            fm.close()
            break

if __name__ == "__main__":
    main()

练习评分标准：

• 代码功能完整性（40%）：实现所有要求的 CRUD 和统计功能。
• 代码健壮性（30%）：包含异常处理、参数化查询。
• 代码可读性（20%）：代码结构清晰，注释适当。
• 用户体验（10%）：命令行界面易用（如果实现）。

4. 扩展学习（可选）

• 学习更复杂的 SQL 查询（如 JOIN、子查询）。
• 探索其他数据库（如 MySQL、PostgreSQL）与 Python 的集成（pymysql、psycopg2）。
• 使用 ORM 框架（如 SQLAlchemy）简化数据库操作。
• 添加数据可视化功能（如使用 matplotlib 绘制收入/支出趋势图）。

教学建议

• 工具准备：确保学员安装 Python 3.x，推荐使用 VS Code 或 PyCharm。
• 教学方式：
• 通过 Jupyter Notebook 或交互式 Python 环境演示代码。
• 提供代码模板，降低初学者上手难度。
• 鼓励学员在练习中自定义类别（如“餐饮”“娱乐”）或添加新功能。
• 答疑重点：
• SQL 语法的常见错误（如引号、字段类型）。
• 数据库连接未关闭导致的资源泄漏。
• 参数化查询的使用场景。

本课程通过理论讲解、代码实践和项目练习，帮助学员掌握 sqlite3 模块的基本操作，并能够将数据库知识应用于实际的财务管理系统开发。学员完成课程后，应能独立设计简单的数据库结构并实现 CRUD 操作，为后续学习更复杂的数据库系统打下基础。