数据库与持久化

学习目标

这一课把内存列表换成真正的数据库：用 sqlite3 建出课程、笔记、学习记录三张有关联的表，跑通迁移、参数化读写、唯一约束和事务，并对照本站 node:sqlite 的用法确认两套 API 是同构的。

问题场景

P08 的笔记存在 NOTES 列表里，服务一重启全没了。你要的是本站那种持久化：data/database/reactlab.db 落盘，重启后学习进度、笔记原样都在。Node 侧本站用的是 Node 24 内置 node:sqlite，Python 侧的标准答案是标准库 sqlite3——同一个 SQLite 引擎，两套驱动 API 几乎逐点对应：prepare/run/get/all 对 execute/fetchone/fetchall。

核心解释

建库脚本一次讲全四件事：迁移、外键、唯一约束、事务。

# db.py
import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS course (
    code TEXT PRIMARY KEY,
    title TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS note (
    id TEXT PRIMARY KEY,
    course_code TEXT REFERENCES course(code),
    title TEXT NOT NULL,
    revision INTEGER NOT NULL DEFAULT 1
);
CREATE TABLE IF NOT EXISTS study_record (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    course_code TEXT NOT NULL,
    minutes INTEGER NOT NULL,
    done INTEGER NOT NULL DEFAULT 0,
    UNIQUE(course_code, done)              -- 演示用：同一课程同状态只允许一条
);
"""

def connect(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute("PRAGMA foreign_keys = ON")   # 默认关闭，必须显式开
    conn.row_factory = sqlite3.Row             # 行变 dict 风格，r["title"] 可用
    conn.executescript(SCHEMA)                 # 迁移：幂等，可重复执行
    return conn

读写与事务：

def add_record(conn: sqlite3.Connection, code: str, minutes: int) -> int:
    try:
        with conn:                             # with = 自动提交/回滚的事务
            cur = conn.execute(
                "INSERT INTO study_record (course_code, minutes) VALUES (?, ?)",
                (code, minutes),               # 参数化：值永远走占位符
            )
            return cur.lastrowid
    except sqlite3.IntegrityError as e:
        raise ValueError(f"重复写入: {e}") from e

四条纪律对应四类事故：

1. 参数化查询是硬规矩。字符串拼接 SQL 是注入入口，? 占位符没有例外——本站 node:sqlite 的 .run(sql, params) 同款约束。
2. 唯一约束交给数据库。应用层「先查再插」在并发下会双双通过；UNIQUE + 捕获 IntegrityError 才是原子防线。
3. 事务边界用 with conn。块内任何异常自动回滚，正常退出自动提交。JS 里没有等价语法糖，容易漏 commit，Python 这层帮你兜住。
4. 迁移幂等。CREATE TABLE IF NOT EXISTS + 版本记录让脚本可重复执行，对应本站 lib/db/migrations 按序执行、记录在 _migrations 表的做法。

示例拆解

验证持久化和约束真的生效：

# main.py
from db import connect, add_record

if __name__ == "__main__":
    conn = connect("study.db")
    conn.execute("INSERT OR IGNORE INTO course VALUES ('P11', '数据库与持久化')")
    rid = add_record(conn, "P11", 30)
    print("写入:", rid)
    row = conn.execute("SELECT course_code, minutes FROM study_record WHERE id=?", (rid,)).fetchone()
    print(dict(row))                            # {'course_code': 'P11', 'minutes': 30}

预期：第一次运行写入成功；删掉脚本重跑，study.db 里的数据还在（持久化）；把 add_record 再跑一次相同参数会抛 ValueError: 重复写入（约束生效）。到 P12 接前端时，这套表结构会直接映射成接口的响应字段。

动手任务

1. 给 note 表加 title 的唯一约束，验证插入重复标题抛 IntegrityError。
2. 写 add_record_with_note：一条学习记录和一条笔记必须在同一事务里写入，人为在第二步抛异常，验证第一步被回滚。
3. 用 PRAGMA foreign_keys 开关对比：给 study_record 插入不存在的 course_code，看两种情况的行为差异。

常见错误

• 忘了 PRAGMA foreign_keys：外键声明了但从不生效，脏数据照进；连接后第一时间开启。
• 手拼 SQL：f"SELECT * FROM note WHERE id={id}" 注入风险直接拉满；一律占位符。
• 连接用完不关：文件锁残留、写入丢失；连接交给 P09 的 yield 依赖或 with 管理。
• autocommit 想当然：默认隔离级别下忘了提交，重启数据「消失」；显式 with conn 管事务边界。

检查与复盘

自查清单：

• 建表脚本可重复执行而不报错（幂等迁移）
• 所有 SQL 值都走 ? 占位符，业务冲突由约束 + IntegrityError 表达
• 能演示一次「事务回滚保护数据一致性」的实际案例

参考资料

• sqlite3 标准库文档 —— 连接、事务与 Row 工厂
• FastAPI 官方教程：SQL Databases —— 服务里组织数据库层的官方路径