23.5 数据库设计 · 波仔的 AI 编程实战课

📄 一句话概括

四张表：三张主体加一张中间表。多对多必须靠中间表，两边都存不了「列表」。 中间表的主键是 (note_id, tag_id)，它顺带保证「同一篇笔记贴同一个标签两次」插不进来。 操作类的价值不是少写代码，是「只有一处能写 SQL」——按人隔离才可能守住。

📝 本章和前面几章的关系

教材 23.5 讲数据库设计：概要说明、创建数据表、数据库操作类。 前两小节常规，第三小节才是重点——把所有 SQL 收进一个类里。 这个类在别处常被当成「代码整洁」的问题，其实它是【安全】问题。 因为 23.1 定的那条规矩「每条 SQL 都要带 user_id」， 只有在一处能写 SQL 的前提下才可能检查。 散落在几十个文件里的 SQL，你没法逐一确认。 演示里量了「封装的开销」——两千次调用，差异落在测量噪声里。 所以「封装会不会拖慢」这个担心基本可以放下。

一、四张表，两种关系

表	存什么	关系
`users`	账号、密码哈希、盐、注册时间	一对多到 `notes`
`notes`	标题、正文、是否删除	多对一到 `users`
`tags`	标签名	多对多到 `notes`
`note_tag`	笔记与标签的对应	中间表，两列外键

一对多和 22 章一样：外键放在「多」的那一边，也就是 notes.user_id。 多对多是新东西：一篇笔记有多个标签，一个标签也挂在多篇笔记上， 两边都不能存「列表」——关系型数据库没有这个类型。 所以中间表不是「可选的优化」，是唯一能表达多对多的方式。 它的每一行就是一句「这篇笔记有那个标签」，没有别的信息。

二、中间表的两个细节

细节一：主键用组合，不用自增 id。 PRIMARY KEY (note_id, tag_id) 这一行同时干了两件事：唯一标识一行， 并且保证同一组合不会出现两次。

写法	能防重复吗	查询效率
`id` 自增主键	不能，要另外加唯一约束	要多一个索引
`(note_id, tag_id)` 组合主键	能	天然带索引

第二行的收益是免费的：组合主键自带一个索引，按 note_id 查很快， 「这篇笔记有哪些标签」一句 SQL 就够。 如果反过来按 tag_id 查得多（「这个标签下有哪些笔记」），就得再加一个索引。 细节二：外键要写 ON DELETE CASCADE。 删笔记的时候，中间表里对应的行必须跟着走，否则会留下「指向不存在笔记」的记录。 演示里删完笔记，中间表剩 0 条，就是这个级联在起作用。

三、操作类：三个方法加一个 insert

教材给的类只有 query 和 execute 两个方法。够用，但有两处可以补：

方法	教材有吗	补它的理由
`query`	有	返回全部行
`query_one`	没有	「取一行」用得太频繁
`execute`	有	返回影响行数
`insert`	没有	要拿自增 id 时必须单独一个方法

第二行是因为「取一行」这个动作在项目里出现的频率极高：数总数、查当前用户、判断存不存在。 每次都写 rows[0] if rows else None 太啰嗦。 第四行是因为 execute 返回的是「影响行数」，拿不到自增 id。 而「插完笔记立刻跳转到它的详情页」这种需求，必须拿到新 id。 取 id 的办法是连接对象上的 lastrowid，所以得单独一个方法。

四、封装到底花不花钱

这是个常被争论的问题：「多一层函数调用会不会慢」。演示里量了 2000 次调用：

写法	2000 次	差异
直接调游标	零点四几秒	—
经操作类	零点四几秒	落在噪声里

结论是：几乎不花钱，而且这个「几乎」是「量不出来」的意思。 因为一次 SQL 查询本身的开销（解析、执行、取行）远大于一次 Python 函数调用。 所以「封装会不会拖慢」这个担心可以放下。真正该担心的是反过来的：不封装， SQL 散落在几十个文件里，以后改表名、加归属条件，就得满仓库找。 而「找漏了」这件事，是不会报错的。

五、动手 Demo：四张表和一个操作类：把 SQL 关进笼子里

这一段建四张表、灌十个用户两千篇笔记，然后用一个操作类走一遍「查一行、插一行、改一行」，再量「封装 vs 直接调」的开销，最后用中间表做一次按标签查笔记。

import sqlite3, time

c = sqlite3.connect(u"db.db", isolation_level=None)
c.execute(u"PRAGMA foreign_keys = ON")
c.executescript(u"""
CREATE TABLE users (id INTEGER PRIMARY KEY, uname TEXT UNIQUE, upass TEXT);
CREATE TABLE notes (id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL,
  title TEXT, is_deleted INTEGER DEFAULT 0);
CREATE TABLE tags (id INTEGER PRIMARY KEY, tname TEXT UNIQUE);
CREATE TABLE note_tag (note_id INTEGER NOT NULL, tag_id INTEGER NOT NULL,
  PRIMARY KEY (note_id, tag_id),
  FOREIGN KEY (note_id) REFERENCES notes(id) ON DELETE CASCADE);
""")
c.execute(u"BEGIN")
c.executemany(u"INSERT INTO users (uname, upass) VALUES (?,?)",
              [(u"user%03d" % i, u"h%d" % i) for i in range(1, 11)])
c.executemany(u"INSERT INTO notes (user_id, title) VALUES (?,?)",
              [(i % 10 + 1, u"笔记%04d" % i) for i in range(1, 2001)])
c.executemany(u"INSERT INTO tags (tname) VALUES (?)",
              [(t,) for t in (u"Python", u"MySQL", u"Flask", u"随笔")])
c.executemany(u"INSERT INTO note_tag VALUES (?,?)",
              [(i, i % 4 + 1) for i in range(1, 2001)])
c.execute(u"COMMIT")

class Db:
    def __init__(self, path):
        self.conn = sqlite3.connect(path, isolation_level=None)
    def query(self, sql, args=()):
        return self.conn.execute(sql.replace(u"%s", u"?"), args).fetchall()
    def query_one(self, sql, args=()):
        r = self.query(sql, args)
        return r[0] if r else None
    def execute(self, sql, args=()):
        return self.conn.execute(sql.replace(u"%s", u"?"), args).rowcount
    def insert(self, sql, args=()):
        return self.conn.execute(sql.replace(u"%s", u"?"), args).lastrowid

db = Db(u"db.db")
print(u"用户 3 有 %d 篇" % db.query_one(
      u"SELECT COUNT(*) FROM notes WHERE user_id = %s", (3,))[0])
nid = db.insert(u"INSERT INTO notes (user_id, title) VALUES (%s,%s)", (3, u"新笔记"))
print(u"插入后 id = %d" % nid)
print(u"改一行影响 %d 行" % db.execute(
      u"UPDATE notes SET title=%s WHERE id=%s AND user_id=%s", (u"改过", nid, 3)))

# 封装的开销
N = 2000
t = time.perf_counter()
for i in range(N):
    c.execute(u"SELECT COUNT(*) FROM notes WHERE user_id = %d" % (i % 10 + 1)).fetchone()
t_raw = time.perf_counter() - t
t = time.perf_counter()
for i in range(N):
    db.query_one(u"SELECT COUNT(*) FROM notes WHERE user_id = %s", (i % 10 + 1,))
t_wrap = time.perf_counter() - t
print(u"直接 %.4f 秒 / 封装 %.4f 秒" % (t_raw, t_wrap))

# 按标签查：连三张表，归属条件照带
rows = db.query(u"SELECT n.id, n.title FROM notes n "
                 u"JOIN note_tag nt ON nt.note_id = n.id "
                 u"JOIN tags t ON t.id = nt.tag_id "
                 u"WHERE t.tname = %s AND n.user_id = %s LIMIT 3", (u"Python", 3))
print(u"按标签查 -> %r" % (rows,))
c.close()

实跑输出：

--------------------------------------------------------------------
四张表和一个操作类：把 SQL 关进笼子里
--------------------------------------------------------------------

**教材 23.5 是数据库设计：概要说明、创建数据表、数据库操作类。**
**前两小节很常规，第三小节才是这一节的重点——把所有 SQL 收进一个类里。**
**这是所有项目的第一层抽象，也是后面「按人隔离」能落地的前提。**

  ① 四张表：三张主体加一张中间表
  --------------------------------------------------------------------
表          存什么              关系
------------------------------------------------
`users`     账号、密码哈希、盐  一对多到 `notes`
`notes`     笔记正文            多对一到 `users`
`tags`      标签字典            多对多到 `notes`
`note_tag`  笔记与标签的对应    中间表，两列外键
**第四张表是这一节唯一的新东西：多对多必须靠中间表。**
**因为一篇笔记可以有多个标签，一个标签也可以挂在多篇笔记上——两边都不能存「列表」。**
**中间表的主键是 `(note_id, tag_id)` 这个组合，它顺带保证了一件事：**
**同一篇笔记贴同一个标签两次，会被主键拦住。**
**这个约束不写代码就有，比在 Python 里判重省事。**

  ② 操作类：四个方法够用
  --------------------------------------------------------------------

  [教材原码 · 本沙箱不执行]
  | # db.py（教材原码）
  | import pymysql
  | 
  | class Db:
  |     def __init__(self):
  |         self.conn = pymysql.connect(host='localhost', user='root',
  |                                     password='123456', database='note',
  |                                     charset='utf8mb4')
  | 
  |     def query(self, sql, args=None):
  |         cur = self.conn.cursor()
  |         cur.execute(sql, args or ())
  |         return cur.fetchall()
  | 
  |     def execute(self, sql, args=None):
  |         cur = self.conn.cursor()
  |         n = cur.execute(sql, args or ())
  |         self.conn.commit()
  |         return n
  教材原码：把连接藏进类里，外面只看到 query 和 execute

query_one      用户 3 有 200 篇
insert         新笔记 id = 2001
execute        影响 1 行

  ③ 封装的代价：多一层函数调用
  --------------------------------------------------------------------
写法               2000 次    每次差多少
----------------------------------------
直接 `c.execute`   0.4474 秒  —
经 `Db.query_one`  0.4457 秒  0.9 微秒
**这次量出来封装版反而略快 0.9 微秒——这说明差异落在测量噪声里，**
**「包一层」这个动作本身几乎不花钱。**
**所以「封装会不会拖慢」这个担心，在绝大多数项目里都是多余的。**
**真正要担心的是反过来的：不封装，SQL 散落在几十个文件里，以后要改表名、要加归属条件，**
**就得满仓库找。**
**第 101 段那个「漏一个条件多看见九百条」，**
**根源就是 SQL 散落——你没法保证每一处都记得加 `user_id`。**

  ④ 按标签找笔记：中间表的典型查询
  --------------------------------------------------------------------
id  title
------------
12  笔记0012
32  笔记0032
52  笔记0052
**这条查询连了三张表，但逻辑简单：从标签名找到标签 id，从中间表找到笔记 id，**
**再从笔记表取内容。**
**注意最后那个 `n.user_id = %s`——即使走了三张表，「按人隔离」依然要带。**
**这就是第 101 段说的「每一个碰笔记表的查询」，中间表连进来的也算。**

这一段里操作类只有四个方法，但已经够覆盖整个项目：query、query_one、execute、 insert。 注意每个方法内部都做了同一件事：把 %s 换成 ?——这是本地复刻的适配层， 真项目里 PyMySQL 直接认 %s，不需要这一行。 「封装 vs 直接调」那两个数放一起看，差异落在噪声里。 这说明「多一层函数调用」的成本相对于一次 SQL 查询可以忽略。 所以封装不封装的判断标准，不是性能，是「以后能不能守住规矩」。 最后那条按标签查的 SQL 值得看结尾：走了三张表，最后仍然带 n.user_id = %s。 这就是「每一个碰笔记表的查询」的实际含义——连从中间表绕进来的也算。 而这类 SQL 在项目里最容易漏，因为它看起来更像「查标签」而不是「查笔记」。

30 秒实验

把 note_tag 的主键去掉，改成自增 id，再插图里同一组合两次，看会不会重复。
删掉 note_tag 的 ON DELETE CASCADE，再删一篇笔记，看报什么错。
给 notes.user_id 建索引，重跑「封装开销」那一段，看两个数是不是都变小。

常见误区

中间表用自增主键。那样挡不住重复插入，还得另外加唯一约束。用 (note_id, tag_id) 组合主键，两件事一起办。
中间表外键不写级联。删笔记会被中间表拦住，或者留下指向不存在笔记的记录。两种都很难查。
担心封装拖慢而把 SQL 到处写。实测差异落在噪声里。真正的代价是「以后没法确认每条 SQL 都带了归属条件」。

自测三问（点开对答案）

1. 为什么多对多必须用中间表？ —— 因为一篇笔记有多个标签、一个标签也挂多篇笔记，两边都不能存「列表」。关系型数据库没有列表类型，只能把这个关系单独存成一张表。 2. 中间表为什么用组合主键而不是自增 id？ —— 组合主键同时干两件事：唯一标识一行，并且保证同一组合不会重复插入。它还天然带一个索引，按 `note_id` 查很快。 3. `insert` 为什么要单独一个方法？ —— 因为 `execute` 返回的是影响行数，拿不到自增 id。而「插完立刻跳详情页」需要新 id，那得从连接的 `lastrowid` 取。

表和操作类都有了，下一篇进用户模块：注册怎么防重复、密码为什么要慢、退出登录到底做了什么。

☐ （本篇完，下一篇 23.6 用户模块设计）