🐍 第22课：数据库操作

—— SQLite：零配置的嵌入式数据库

🏆 CRUD+事务+记账本

✅ Python验证通过

📌 本课目标

掌握 SQLite 数据库创建与连接
掌握建表、插入、查询、更新、删除（CRUD）
掌握参数化查询防 SQL 注入
掌握事务控制（BEGIN/COMMIT/ROLLBACK）
掌握 sqlite3.Row 与连接管理最佳实践

1️⃣ SQLite 基础

import sqlite3

# 连接数据库（不存在则创建）
conn = sqlite3.connect("myapp.db")

# 内存数据库（测试用，进程结束即消失）
conn = sqlite3.connect(":memory:")

# 创建游标
cursor = conn.cursor()

# 建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    age INTEGER DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")

# 创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_email ON users(email)")
conn.commit()

💡 SQLite 的优势：零配置、单文件、ACID 事务、支持大多数 SQL 标准。适合中小应用、嵌入式场景、数据分析。并发写入有限（单写者锁），读并发无限制。

2️⃣ CRUD 操作

import sqlite3

conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

# 建表
cursor.execute("""
CREATE TABLE products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    price REAL NOT NULL,
    stock INTEGER DEFAULT 0,
    category TEXT
)
""")

# ========== INSERT ==========
# 单条插入（参数化查询！）
cursor.execute(
    "INSERT INTO products (name, price, stock, category) VALUES (?, ?, ?, ?)",
    ("MacBook Pro", 14999.0, 50, "电脑")
)

# 批量插入（高效）
products = [
    ("iPhone 15", 7999.0, 200, "手机"),
    ("AirPods Pro", 1899.0, 500, "配件"),
    ("iPad Air", 4799.0, 100, "平板"),
    ("Apple Watch", 2999.0, 150, "配件"),
]
cursor.executemany(
    "INSERT INTO products (name, price, stock, category) VALUES (?, ?, ?, ?)",
    products
)
conn.commit()
print(f"插入了 {cursor.rowcount} 条记录")

# ========== SELECT ==========
# 查询所有
cursor.execute("SELECT * FROM products")
for row in cursor.fetchall():
    print(row)

# 条件查询
cursor.execute("SELECT name, price FROM products WHERE price > ?", (3000,))
expensive = cursor.fetchall()

# 模糊查询
cursor.execute("SELECT * FROM products WHERE name LIKE ?", ("%Pro%",))

# 排序和限制
cursor.execute("SELECT * FROM products ORDER BY price DESC LIMIT 3")

# 聚合查询
cursor.execute("SELECT category, COUNT(*) as cnt, AVG(price) as avg_price FROM products GROUP BY category")

# ========== UPDATE ==========
cursor.execute("UPDATE products SET price = ? WHERE name = ?", (13999.0, "MacBook Pro"))
conn.commit()
print(f"更新了 {cursor.rowcount} 条记录")

# ========== DELETE ==========
cursor.execute("DELETE FROM products WHERE stock = ?", (0,))
conn.commit()

⚠️ 永远不要用 f-string 或字符串拼接构造 SQL！这会导致 SQL 注入漏洞。始终使用 ? 参数化查询。

3️⃣ 事务控制

import sqlite3

conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

cursor.execute("""
CREATE TABLE accounts (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    balance REAL NOT NULL DEFAULT 0
)
""")
cursor.executemany("INSERT INTO accounts (name, balance) VALUES (?, ?)",
    [("Alice", 1000.0), ("Bob", 500.0)])
conn.commit()

# 转账事务：A → B 转 200
def transfer(conn, from_name, to_name, amount):
    """事务性转账"""
    try:
        cursor = conn.cursor()
        # 检查余额
        cursor.execute("SELECT balance FROM accounts WHERE name = ?", (from_name,))
        result = cursor.fetchone()
        if not result or result[0] < amount:
            raise ValueError(f"{from_name} 余额不足")
        # 扣款
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?",
                      (amount, from_name))
        # 加款
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?",
                      (amount, to_name))
        conn.commit()
        print(f"✅ 转账成功: {from_name} → {to_name} ¥{amount}")
    except Exception as e:
        conn.rollback()
        print(f"❌ 转账失败: {e}")

# 测试
transfer(conn, "Alice", "Bob", 200)
cursor.execute("SELECT name, balance FROM accounts")
for row in cursor:
    print(f"  {row[0]}: ¥{row[1]}")

4️⃣ sqlite3.Row 与上下文管理

import sqlite3
from contextlib import contextmanager

# 使用 Row 工厂，支持列名访问
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row

cursor = conn.cursor()
cursor.execute("""
CREATE TABLE logs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    level TEXT,
    message TEXT,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.executemany("INSERT INTO logs (level, message) VALUES (?, ?)", [
    ("INFO", "服务启动"),
    ("WARNING", "内存使用率80%"),
    ("ERROR", "数据库连接超时"),
])
conn.commit()

# 用列名访问
cursor.execute("SELECT * FROM logs")
for row in cursor:
    print(f"[{row['level']}] {row['message']} ({row['timestamp']})")

# 转为字典
cursor.execute("SELECT * FROM logs WHERE level = ?", ("ERROR",))
error_logs = [dict(row) for row in cursor.fetchall()]

# 上下文管理器封装
@contextmanager
def get_db(db_path=":memory:"):
    """数据库连接上下文管理器"""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# 使用
with get_db() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) as cnt FROM logs")
    print(f"日志总数: {cursor.fetchone()['cnt']}")

5️⃣ 实战：简易记账本

import sqlite3
from datetime import datetime

class AccountBook:
    """SQLite 实现的记账本"""
    
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._init_db()
    
    def _init_db(self):
        self.conn.execute("""
        CREATE TABLE IF NOT EXISTS records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            type TEXT NOT NULL CHECK(type IN ('income', 'expense')),
            category TEXT NOT NULL,
            amount REAL NOT NULL CHECK(amount > 0),
            note TEXT DEFAULT '',
            date TEXT NOT NULL
        )""")
        self.conn.commit()
    
    def add(self, type_, category, amount, note="", date=None):
        date = date or datetime.now().strftime("%Y-%m-%d")
        self.conn.execute(
            "INSERT INTO records (type, category, amount, note, date) VALUES (?, ?, ?, ?, ?)",
            (type_, category, amount, note, date)
        )
        self.conn.commit()
        print(f"✅ 记录: {type_} {category} ¥{amount:.2f}")
    
    def summary(self):
        cursor = self.conn.execute(
            "SELECT type, SUM(amount) as total FROM records GROUP BY type"
        )
        result = {row["type"]: row["total"] for row in cursor}
        income = result.get("income", 0)
        expense = result.get("expense", 0)
        print(f"收入: ¥{income:.2f}  支出: ¥{expense:.2f}  余额: ¥{income - expense:.2f}")
    
    def category_breakdown(self):
        cursor = self.conn.execute("""
        SELECT category, type, SUM(amount) as total 
        FROM records GROUP BY category, type ORDER BY total DESC
        """)
        for row in cursor:
            emoji = "📈" if row["type"] == "income" else "📉"
            print(f"  {emoji} {row['category']}: ¥{row['total']:.2f}")
    
    def close(self):
        self.conn.close()

# 使用
book = AccountBook()
book.add("income", "工资", 15000)
book.add("expense", "房租", 3000)
book.add("expense", "餐饮", 1500, "外卖+食堂")
book.add("income", "副业", 2000, "接外包")
book.add("expense", "购物", 800)
book.summary()
book.category_breakdown()
book.close()

6️⃣ 验证脚本

#!/usr/bin/env python3
"""第22课 SQLite CRUD验证"""
import sqlite3
import tempfile
import os

def test_crud():
    """CRUD全流程测试"""
    conn = sqlite3.connect(":memory:")
    c = conn.cursor()
    c.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, value REAL)")
    c.execute("INSERT INTO test (name, value) VALUES (?, ?)", ("a", 1.5))
    c.executemany("INSERT INTO test (name, value) VALUES (?, ?)",
                  [("b", 2.5), ("c", 3.5)])
    conn.commit()
    c.execute("SELECT COUNT(*) FROM test")
    assert c.fetchone()[0] == 3
    c.execute("UPDATE test SET value = ? WHERE name = ?", (99.0, "a"))
    conn.commit()
    c.execute("DELETE FROM test WHERE name = ?", ("c",))
    conn.commit()
    c.execute("SELECT COUNT(*) FROM test")
    assert c.fetchone()[0] == 2
    conn.close()
    print("✅ CRUD全流程测试通过")

def test_transaction():
    """事务测试"""
    conn = sqlite3.connect(":memory:")
    c = conn.cursor()
    c.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
    c.execute("INSERT INTO t (v) VALUES (1)")
    conn.commit()
    try:
        c.execute("INSERT INTO t (v) VALUES (2)")
        raise RuntimeError("模拟错误")
    except:
        conn.rollback()
    c.execute("SELECT COUNT(*) FROM t")
    assert c.fetchone()[0] == 1
    conn.close()
    print("✅ 事务测试通过")

def test_row_factory():
    """Row工厂测试"""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    c = conn.cursor()
    c.execute("CREATE TABLE r (id INTEGER PRIMARY KEY, name TEXT)")
    c.execute("INSERT INTO r (name) VALUES ('test')")
    conn.commit()
    c.execute("SELECT * FROM r")
    row = c.fetchone()
    assert row["name"] == "test"
    assert dict(row)["name"] == "test"
    conn.close()
    print("✅ Row工厂测试通过")

def test_file_persistence():
    """文件持久化测试"""
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = os.path.join(tmpdir, "test.db")
        conn = sqlite3.connect(db_path)
        conn.execute("CREATE TABLE p (v TEXT)")
        conn.execute("INSERT INTO p (v) VALUES ('persist')")
        conn.commit()
        conn.close()
        conn2 = sqlite3.connect(db_path)
        result = conn2.execute("SELECT v FROM p").fetchone()
        assert result[0] == "persist"
        conn2.close()
    print("✅ 文件持久化测试通过")

if __name__ == "__main__":
    test_crud()
    test_transaction()
    test_row_factory()
    test_file_persistence()
    print("\n🎉 第22课全部验证通过！")

7️⃣ SQL 注入防护

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, password TEXT)")
conn.execute("INSERT INTO users (name, password) VALUES ('admin', 'secret123')")
conn.commit()

# ❌ 危险：字符串拼接
# username = "admin' OR '1'='1"
# cursor.execute(f"SELECT * FROM users WHERE name = '{username}'")
# 这会返回所有行！

# ✅ 安全：参数化查询
username = "admin"
cursor = conn.execute("SELECT * FROM users WHERE name = ?", (username,))
print(f"结果: {cursor.fetchall()}")

# 命名占位符
conn.execute("SELECT * FROM users WHERE name = :name AND id > :min_id", 
             {"name": "admin", "min_id": 0})

⚠️ SQL 注入是最常见的 Web 安全漏洞之一。永远不要信任用户输入，始终使用参数化查询。

8️⃣ SQLite 高级技巧

import sqlite3

conn = sqlite3.connect(":memory:")

# 1. WAL 模式（写前日志）——提升并发读性能
conn.execute("PRAGMA journal_mode=WAL")

# 2. 批量操作优化 ——事务包裹
conn.execute("BEGIN TRANSACTION")
for i in range(10000):
    conn.execute("INSERT INTO t VALUES (?)", (i,))
conn.execute("COMMIT")
# 比逐条 autocommit 快 10-100 倍

# 3. UPSERT（INSERT OR REPLACE）
conn.execute("""
CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT)
""")
conn.execute("INSERT OR REPLACE INTO kv (key, value) VALUES (?, ?)", 
             ("config", "new_value"))

# 4. 常用 PRAGMA
conn.execute("PRAGMA cache_size = -64000")   # 64MB 缓存
conn.execute("PRAGMA synchronous = NORMAL")   # 性能折中
conn.execute("PRAGMA busy_timeout = 5000")   # 忙等待5秒

💡 SQLite 不适合的场景：高并发写入（>1写/秒）、多进程写、需要复杂权限控制、TB级数据。这些场景请用 PostgreSQL/MySQL。

9️⃣ SQLite 与 Pandas

import sqlite3
import pandas as pd

conn = sqlite3.connect(":memory:")

# 从 CSV 导入到 SQLite
# df = pd.read_csv("data.csv")
# df.to_sql("my_table", conn, if_exists="replace", index=False)

# 创建测试数据
conn.execute("CREATE TABLE sales (id INTEGER PRIMARY KEY, product TEXT, amount REAL, date TEXT)")
conn.executemany("INSERT INTO sales (product, amount, date) VALUES (?, ?, ?)",
    [("手机", 3999, "2024-01-15"), ("电脑", 7999, "2024-01-16"),
     ("耳机", 899, "2024-01-15"), ("平板", 2999, "2024-01-17"),
     ("手机", 4299, "2024-01-18"), ("电脑", 6999, "2024-01-19")])
conn.commit()

# SQLite → DataFrame
df = pd.read_sql("SELECT * FROM sales", conn)
print(df)

# 聚合查询 → DataFrame
summary = pd.read_sql("""
    SELECT product, COUNT(*) as count, SUM(amount) as total, AVG(amount) as avg
    FROM sales GROUP BY product ORDER BY total DESC
""", conn)
print(summary)

# DataFrame → SQLite
summary.to_sql("summary", conn, if_exists="replace", index=False)

# 分页查询
page_size = 10
page = 1
offset = (page - 1) * page_size
paged = pd.read_sql(f"SELECT * FROM sales LIMIT {page_size} OFFSET {offset}", conn)

🔑 本课要点

sqlite3——Python 内置，零配置，单文件数据库
参数化查询——用 ? 占位符，永远不要用 f-string 拼 SQL
事务控制——BEGIN/COMMIT/ROLLBACK 保证数据一致性
连接管理——使用 context manager 自动管理连接
Row 工厂——sqlite3.Row 让结果支持列名访问

← 上一课 📚 目录下一课 →