← 代码课堂

第 04 课：数据层 —— SQLite 封装与表设计

难度：★★☆（需要会 Python 函数和字典）
教材：查资料/server.py 的数据层（第 472-569 行 + 第 1086-1246 行）
预计时间：讲解 60 分钟 + 练习 40 分钟
学完你能：看懂一张表为什么这么设计，会写安全的增删改查，理解索引、事务、并发锁

---

主线实验：一张卡片何时才算真正保存

前置：能读 Python 函数、列表字典和异常，已完成 02 的一次请求。先在下载包 assets/主线实验/ 运行 server.py 并创建卡片，停止服务后再启动、刷新，卡片应仍在。

本课主问题是：怎样把经过校验的 title 写入数据库，再作为同一条记录读回来？ 表设计、参数化查询和事务都围绕这个问题展开；全文检索、ORM 和连接池放到第二遍选读。

写入链路的四个明确阶段

1. validate_title 检查空值与长度，失败时尚未执行 INSERT。
2. connect 创建本次操作自己的数据库连接；连接指向 --db 选定的文件。
3. execute 使用 ? 传入 title，数据库把它当参数值，而不是拼接后的 SQL 语法。
4. with conn 正常退出时提交，异常退出时回滚；外面的 closing 负责关闭连接。

不要把 with conn 误读为“退出就关闭连接”。它管理事务，关闭连接是另一项资源管理工作。也不要把“没有执行 commit”说成异常瞬间必然已经完成 rollback；应看明确的上下文管理或回滚调用。

先观察参数化，再观察事务

通过页面保存标题 学习 'SQL' <标签>。应当原样保存并以文本显示，不因为单引号改变 SQL。参数化针对值的绑定，不代表可以把任意用户输入当表名、列名或 SQL 片段。

运行 py -3 transaction_demo.py。它在内存表中先插入 25，再插入违反 CHECK 约束的 -1。预期打印“事务已回滚”和 回滚后记录数：0。第二步出错后，两步都没有成为已提交记录。

这个实验使用同一个事务。若你故意在第一步后先 commit，结果会不同：第一步已成为独立提交，后面的失败不能撤销它。先预测，再在自己的副本中验证。

持久化与共享是两件事

SQLite 能保存结构化数据并支持查询；它不会自动给另一台电脑提供访问接口。多端共享通常是多个客户端访问同一个服务，由服务操作数据库并检查权限。每台设备各有一份 SQLite 文件仍然是多份独立数据。

WAL 能改善读写并发，但 SQLite 仍通常只有一个写事务在提交，锁等待仍可能发生。check_same_thread=False 仅关闭同连接跨线程的保护，不会自动使共享连接安全；每次操作自己创建连接时通常不需要它。

练习与验收

先交付三项证据：特殊标题写入并读回；空白标题拒绝且数量不变；两条事务第二条失败后总数为 0。再做原课 rating 扩展，明确新建表和已有库迁移是两条不同路径。

排错：no such table 先检查是否连到了错误的新空文件；database is locked 检查其他连接和未结束事务；数据消失先打印数据库绝对路径。需要备份活动数据库时用 SQLite backup API；尤其 WAL 活跃时不能只复制主文件就保证完整。

---

课前须知：回答上节的思考题

第 03 课练习 4 问：两台电脑想同步数据，localStorage 行不行？答案是不行。
localStorage 只属于"这台电脑的这个浏览器"：

• 换电脑打开 → 数据不在
• 清除该站点的存储数据 → localStorage 可能被删除（普通资源缓存清理不一定删除它）
• 手机和电脑同时用 → 各存各的

需要共享数据时，通常由服务端提供统一接口，再由它访问数据库；数据库文件本身不会自动让所有设备共享，也不应让任意人直接读写。
「查资料」用的就是 SQLite——世界上用得最多的数据库（你的手机、浏览器里都有它），
带两个巨大好处：

1. 不用安装任何数据库软件，Python 自带支持
2. 数据持久化到本地数据库（knowledge.db）；备份活动数据库可用 backup API，WAL 活跃时不能只复制主文件就假定完整

---

第一步：亲眼看数据

方式 A（推荐）：装个可视化工具

搜索下载 DB Browser for SQLite（免费开源），用它打开
查资料/knowledge.db，切到 "Browse Data" 标签，
你就能像看 Excel 一样看到 cards 表和 links 表。

方式 B：用 Python 直接看（不用装东西）

在项目目录打开命令行，输入 py -3 进入交互模式，然后：

import sqlite3
conn = sqlite3.connect("knowledge.db")
conn.execute("SELECT id, title, tags, created_at FROM cards").fetchall()

你会看到每一张卡片是一条"元组"。再试试：

conn.execute("SELECT a, b FROM links").fetchall()   # 看卡片之间的关系

观察实验

1. 在网页上新建一张卡片，再执行上面的查询 → 多了一行
2. 把卡片移出收件箱，再查一次 → inbox 字段从 1 变成 0
3. 想一想：你在网页上点的每一个按钮，最后都变成了对这张表的读写——这就是数据层的意义。

---

第二步：模块地图（数据层）

数据层不是一个文件，而是散在 server.py 里的一组"约定"：

数据层（server.py）
│
├── 连接：db()                 第 475-480 行   开连接 + 统一设置
├── 建表：init_db()            第 483-509 行   定义表结构（启动时跑一次）
├── 翻译：row_to_card()        第 516-528 行   数据库行 → Python 字典
├── 规则：validate_sources()   第 555-569 行   数据入库前的校验
├── 并发：_db_lock             第 472 行       多线程写入的"排队锁"
│
└── 使用它的网络层方法（复习第 02 课：业务方法）
      _card_create   第 1092 行   增
      _card_action   第 1122 行   改（link / move / set-tags / dup）
      _card_delete   第 1176 行   删
      fulltext_search 第 720 行   查（关键词）
      _graph         第 1229 行   查（拼图谱数据）

规律：数据层提供"原语"（连接、建表、翻译），业务层负责"组合"它们完成需求。

---

第三步：逐块精讲

块 1：db() —— 每次操作开一个连接（第 475-480 行）

def db():
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn

四行设置，每行都有讲究：

设置	人话解释	不设置会怎样
check_same_thread=False	关闭同连接跨线程检查，不自动保证线程安全	只有跨线程复用同一个连接时才涉及；各线程自建连接通常不需要关闭
row_factory = sqlite3.Row	查询结果可以按列名取：row["title"]	只能按序号取：row[1]，难读易错
PRAGMA journal_mode=WAL	换一种日志模式，读和写能同时进行	一边写一边读时容易报"database is locked"
PRAGMA foreign_keys=ON	打开外键约束检查	SQLite 默认是关闭的，约束形同虚设

注意这个项目的做法是：每次要用就 conn = db()，用完 conn.close()。
简单、不容易出错（代价是频繁开关有开销）。大项目的优化方向是"连接池"，但那是后话。

块 2：init_db() —— 表设计（第 483-509 行）

这是本课最该看懂的代码，完整贴出来：

CREATE TABLE IF NOT EXISTS cards (
    id         TEXT PRIMARY KEY,
    title      TEXT NOT NULL,
    tags       TEXT NOT NULL DEFAULT '[]',
    summary    TEXT NOT NULL DEFAULT '',
    body       TEXT NOT NULL DEFAULT '',
    sources    TEXT NOT NULL DEFAULT '[]',
    inbox      INTEGER NOT NULL DEFAULT 1,
    question   TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS links (
    a TEXT NOT NULL,
    b TEXT NOT NULL,
    PRIMARY KEY (a, b)
);
CREATE INDEX IF NOT EXISTS idx_cards_created ON cards(created_at);
CREATE INDEX IF NOT EXISTS idx_cards_inbox  ON cards(inbox);

逐个设计点讲解（这是"读表结构"的通用方法）：

1. id 用 TEXT 而不是自增数字。

看第 1094 行：cid = uuid.uuid4().hex[:12]——用随机 12 位十六进制。
好处：减少集中分配 ID 的需要；但截短随机 ID 仍可能碰撞，需要唯一约束与冲突处理，不能保证多设备合并绝不冲突。

2. tags 和 sources 存的是 JSON 文本（'[]'、'{}'）。

SQLite 没有"数组"这种类型，所以把列表打包成字符串。
简单项目够用，但代价是普通索引和约束不如规范化关联表直接；支持 JSON 函数时仍可按标签查询——见练习 3。

3. inbox 用 INTEGER 存 0/1。

SQLite 没有布尔类型，用整数代替；读出来时在 row_to_card() 里转回 True/False。

4. 时间存成 TEXT（"2026-09-13 10:30:00"）。

为什么不用时间类型？因为这种固定格式的文本排序结果和真实时间一致，
而且人一眼能看懂，调试方便。

5. links 表用 (a, b) 当复合主键。

一行代表"a 和 b 有关系"。复合主键天然去重——同样的 a、b 插两次会失败，
所以代码里用 INSERT OR IGNORE（第 1117 行）"插了重复的就忽略"。

6. 两个索引 idx_cards_created / idx_cards_inbox。

索引就像书的目录：没有目录，找内容要一页页翻（全表扫描）；
有目录，直接跳过去。这两个字段经常出现在 WHERE / ORDER BY 里，所以建索引。

7. CREATE TABLE IF NOT EXISTS：重复运行不出错（幂等），

所以启动时无脑调用就行。

块 3：row_to_card() —— 数据库和程序的"翻译官"（第 516-528 行）

def row_to_card(row):
    return {
        "id": row["id"],
        "title": row["title"],
        "tags": json.loads(row["tags"]),      # JSON 文本 → Python 列表
        "sources": json.loads(row["sources"]),
        "inbox": bool(row["inbox"]),          # 0/1 → False/True
        ...
    }

为什么要有这一层？因为数据库的存储格式 ≠ 接口需要的格式：

• 数据库里 tags 是字符串 '["天文","物理"]'
• 接口要返回的是真正的数组 ["天文","物理"]

这层转换把"存储细节"包在数据层里。以后就算把 SQLite 换成别的数据库，
能减少存储格式变化对调用方的影响；真正更换数据库时，连接、SQL 方言、事务和迁移仍可能要一起调整。这就是封装的收益。

块 4：写入的三道保险（_card_create()，第 1092-1120 行）

新增一张卡片的完整流程，重点看三个安全设计：

保险一：参数化查询（防 SQL 注入）

conn.execute(
    "INSERT INTO cards (...) VALUES (?,?,?,?,?,?,?,?,?,?)",
    (cid, title, ...),
)

永远不要这样拼 SQL：

# 危险写法（千万不要学）
sql = f"INSERT INTO cards (title) VALUES ('{user_title}')"

把输入拼成 SQL 可能改变语句含义；这个多语句样例在 sqlite3.execute 中通常会被拒绝，不应承诺一定删表。参数化绑定让值与 SQL 结构分离，适合这里的标题字段；动态表名、列名仍需单独限制。

保险二：事务（要么全成功，要么全不动）

conn.execute("INSERT INTO cards ...")          # 第 1 步：插入卡片
for lid in data.get("links") or []:            # 第 2 步：插入它和其他卡片的关联
    conn.execute("INSERT OR IGNORE INTO links ...")
conn.commit()                                   # 到这里才真正"落盘"

若第 2 步失败，应由事务上下文或显式 rollback 回滚；仅仅出现异常不代表所有连接都已立即完成回滚。主线实验用 with conn 展示这一保证。
不会出现"卡片建了但关系没建"的半成品状态。

保险三：线程锁 _db_lock（第 472 行定义，第 1098 行使用）

_db_lock = threading.Lock()
...
with _db_lock:
    conn.execute(...)
    conn.commit()

还记得第 02 课吗？服务器是 ThreadingHTTPServer，每个请求一个线程。
两个人都点了"保存"，项目若每次操作各自建连接，就不是两个线程在共享同一个连接；它们仍可能竞争数据库写锁。
锁的作用就是"写入排队，一次只进一个"。
（WAL 改善读写并发，但仍有事务隔离、锁等待和一致性边界；不要推导成任何读取都不会等待。）

块 5：改与删（第 1122-1183 行）

_card_action() 用第 02 课学过的表驱动分派动作：link（加关系）、
unlink（删关系）、move（进出收件箱）、set-tags（改标签）、dup（复制卡片）。

两个值得注意的细节：

1. unlink 要删两次方向（第 1137 行）：

WHERE (a=? AND b=?) OR (a=? AND b=?)
因为 links 存的是"无向关系"，a→b 和 b→a 都代表同一条关系。

2. 删卡片时先删关系再删卡片（第 1179-1180 行）。

links 表没有建外键约束，所以"删卡片时要顺手清理它的关系"这件事，
得靠代码手动做。如果建了外键 + ON DELETE CASCADE，数据库会自动帮你删——
这是表设计上可以改进的地方（练习里会让你思考）。

块 6：查询的两种思路（第 720-749 行 + 1229 行）

关键词搜索用 SQL（fulltext_search，第 720 行）：

like = "%" + q + "%"
rows = conn.execute(
    "SELECT * FROM cards WHERE title LIKE ? OR body LIKE ? ...",
    (like, like, like, like),
)

% 是通配符，%天文% 表示"任意位置包含天文"。
注意：% 开头的 LIKE 用不上索引，数据量一大就慢——这是 SQL 的局限。

语义搜索拉回内存算（semantic_search，第 732 行）：
它先把最多 500 张卡片全部取出来，再用 Python 算 TF-IDF 和余弦相似度。
注释写着"接口预留向量库接入点"——作者很清楚将来数据量大了要换方案。

教学点：数据库擅长"按条件筛选、排序、统计"，不擅长复杂计算。
能用 SQL 解决的交给 SQL，算不动的拉回程序里算，这是很实用的分工直觉。

---

第四步：对照开源，别人怎么写数据层

对照 sqlite3 官方文档

这些 API 都可在官方文档中查到，但选项要结合连接归属、事务与备份场景判断；真实项目的历史实现也存在可改进之处。
去 Python 官网文档搜 sqlite3，你会发现示例和教材里的写法一一对应。

对照 ORM：SQLAlchemy / Django ORM

项目现在写的是"手写 SQL"：拼 SQL 字符串 + 手动转换结果。
业界另一种主流做法叫 ORM（对象关系映射），把表映射成类：

# 用 SQLAlchemy 的话，可能是这样（示意）
class Card(Base):
    __tablename__ = "cards"
    id = Column(String, primary_key=True)
    title = Column(String)

# 查询变成操作对象，而不是写 SQL
cards = session.query(Card).filter(Card.title.like("%天文%")).all()

	手写 SQL（教材项目）	ORM
上手理解	直观，能看懂每条 SQL	需要先学框架
写起来	字段多了比较啰嗦	快，少写重复代码
防注入	靠你记得用 ?	框架自动处理
复杂查询	完全掌控	有时会别扭
换数据库	要改 SQL 方言	基本不用改

学习顺序建议：先手写（看懂原理，你已经做到了），再学 ORM（提高效率）。
反过来先学 ORM，容易变成"只会调 API 不懂数据库"。

对照更大型的数据库

SQLite 是"嵌入式数据库"：无服务、单文件、适合本地应用。
如果你的「查资料」将来要给很多人在线用，会换成 MySQL/PostgreSQL：
它们是"客户端-服务器"模式，能扛更高并发。
好消息是：SQL 语句和表设计思路基本通用，你的数据层结构不用大改。

对照你的另一个项目

etymology-agent（词源机器人）也用 SQLite 存本地词源缓存——
同一个技术，不同场景。等你上完这课，可以回去看看它的建表语句，对比一下。

---

动手练习（做完才算过关）

练习 1（热身）：命令行查数据
用上面"方式 B"进入 py -3 交互模式，查出：
（1）一共有几张卡片 （2）inbox=1（收件箱）的有几张
提示：SELECT COUNT(*) FROM cards、WHERE inbox=1。

练习 2（核心）：给卡片加一个"评分"字段
要求：卡片能存一个 0-5 的评分。你需要动 4 个地方，找到它们：
（1）init_db() 里加列（新库生效）
（2）对已经存在的 knowledge.db 执行一次 ALTER TABLE（老库生效）：

ALTER TABLE cards ADD COLUMN rating INTEGER NOT NULL DEFAULT 0;

（3）row_to_card() 里把新字段翻译出来
（4）_card_create() 的 INSERT 里带上它
体会一下："数据库加一个字段，代码要动几处"——这就是表设计要慎重的原因。

练习 3（体会 JSON 列的代价）：
试着用 SQL 找出所有带"天文"标签的卡片。一种简单但依赖 JSON 文本表示的尝试是：

SELECT * FROM cards WHERE tags LIKE '%"天文"%';

然后思考：如果一张卡片的标签是"天文观测"，会被搜出来吗（会/不会，为什么）？
更好的设计是什么？（提示：单独建一张 card_tags(card_id, tag) 关联表，或用 SQLite 的 JSON1 函数）

练习 4（思考题）：
fulltext_search 用的是 %关键词%，为什么它用不上索引？
如果卡片涨到 10 万张、搜索变慢，你会怎么优化？

看答案

SQLite FTS5 全文索引——这是标准答案，查一下它是什么

练习 5（选做，练"讲明白"）：
给 _db_lock 那一行写一句注释，要求：一个完全不懂编程的人也能看懂为什么需要它。
写完自己读一遍：一个完全不懂编程的人能看懂吗？

---

本课小结

• SQLite 提供结构化查询和事务；跨设备访问需要服务接口，持久化保证还取决于事务、存储与备份配置
• 表设计要看懂五件事：主键、字段类型、默认值、关系怎么表达、索引建在哪
• db() 的四个设置、row_to_card() 的翻译层，都是"封装"思想
• 写数据三道保险：参数化查询防注入、事务保一致、线程锁防并发
• 查询分工：SQL 干筛选排序，复杂计算拉回程序
• 先手写 SQL 看懂原理，再上 ORM 提效率——顺序不要反

术语表

术语	人话解释
数据库 / 表 / 字段 / 行	仓库 / 表格 / 列 / 一条记录
主键（PRIMARY KEY）	每一行的唯一身份证
外键（FOREIGN KEY）	指向另一张表的"引用"，用来保证关系有效
索引（INDEX）	加速查找的目录；代价是占空间、写入稍慢
事务（transaction）	一组操作打包，要么全成要么全废
commit	提交事务使改动成为已提交状态；底层落盘细节取决于日志和同步配置
SQL 注入	把恶意 SQL 混进用户输入来搞破坏；用 ? 参数化查询防住
参数化查询	用占位符传值，让数据库自己安全地替换
全表扫描	没有索引时一行行翻，慢
WAL	SQLite 的一种日志模式，让读写能并行
ORM	用"操作对象"代替"写 SQL"的框架

下节预告

第 05 课：前端三件套：浏览器怎么和后端说话。
这次教材换成 查资料/web/ 文件夹（index.html / mobile.html + css + js）。
我们会讲：网页怎么发请求（fetch）、拿到的 JSON 怎么变成页面、
以及"同一个后端如何服务电脑端和手机端两套界面"——
上完这课，一个完整项目的前后端你就全打通了。