2026.11.01 AI学习

AI编程（五十五）：Python+SQLite — 让代码亲手接管数据库

作者：climbing

前几篇的代码里，connect、cursor、execute、commit 已经用得飞起，但它们到底是干嘛的？为什么要 commit？忘了 close 会怎样？SQL 语句里拼字符串有什么致命危险？今天不学新 SQL，专门把 Python 和数据库之间那根”电话线”——sqlite3 模块——从头到尾理一遍，顺手打上”防注入”的疫苗。从此你写的数据库代码，才算真正上了”正经软件”的秤。

一、真实场景：手动敲 SQL 敲到手抽筋

接着价格监控工具的故事讲。JOIN 学完之后，你有了新想法：让程序每天自动把新查到的价格写进数据库，别再让自己开命令行、手动敲 INSERT INTO ... 敲到手抽筋。

其实从第 50 篇起，我们一直在用 Python 操作数据库——只是当时光顾着学 SQL，没顾上细看”Python 这边”的家伙事儿。今天换个主角：SQL 只当配角，主角是 sqlite3 这个标准库模块，也就是 Python 和 SQLite 数据库之间那根”电话线”。

打个比方：数据库是一家档案馆，你的 Python 程序是外面的办事员。你不能隔空喊话让档案馆干活，得走一套标准流程：开专线、拿话筒、喊指令、等盖章、挂电话。今天就把这套流程的每一步认全——听起来琐碎，但90% 的”数据怎么没存上””程序怎么卡住了””怎么被人黑了”的事故，都出在这套流程没走对。

二、四件套登场：connect、cursor、execute、commit

先看一段最小可用的模板，所有 Python 操作数据库的代码，骨架都长这样：

# template.py — Python 操作数据库的最小骨架
import sqlite3

# 第1步 connect：开一条通往档案馆的专线
conn = sqlite3.connect("shop.db")

# 第2步 cursor：拿起话筒——真正干活的是它
cur = conn.cursor()

# 第3步 execute：喊一条指令
cur.execute("SELECT name, brand FROM products")

# 第4步 fetch：把查到的档案复印带回家
rows = cur.fetchall()
for row in rows:
    print(row)

# 第5步 close：挂电话
conn.close()

四件套各管一摊，用档案馆的比方串起来：

  • connect（连接）：sqlite3.connect("shop.db")——开一条专线。文件不存在就自动新建，所以”连一个不存在的库”不报错，第一次跑完凭空多出一个 .db 文件，别慌，这是正常行为。
  • cursor（游标）：拿起话筒。它是”真正干活的手”：execute、fetchall 都是它的动作。为什么非要多这一层？因为设计上一条连接可以有多个游标、同时挂着多个查询——单线程小脚本用不上这个本事，但话筒该拿还得拿。
  • execute / executemany（执行）：对档案馆喊一条指令（execute），或者一口气喊一摞——executemany 批量插入 N 条数据，比循环里调 N 次 execute 快得多，这是它的独门本事。
  • commit（提交）：盖章确认。增、删、改这类”动数据”的操作，喊完只是”说了一嘴”，档案馆记在便签上等着；conn.commit() 盖了章，才算正式入档。查数据（SELECT）不用盖章——看档案不改档案。
  • fetchone / fetchall / fetchmany（取结果）：把查询结果复印带回家。fetchone 取一行，fetchall 取全部，fetchmany(n) 取 n 行。
  • close（关闭）：挂电话。专线是系统资源，不挂就一直占着。

三、增删改查全流程：把上几篇的活儿重做一遍

下面这份代码从建表到查数据一条龙，完整可运行——不用敲 SQL 命令行，全靠 Python：

# crud.py — 用 Python 完成增删改查全流程
import sqlite3

conn = sqlite3.connect("shop.db")
cur = conn.cursor()

# 建表（IF NOT EXISTS：有则跳过，程序可以反复跑）
cur.execute("""
CREATE TABLE IF NOT EXISTS products (
    id    INTEGER PRIMARY KEY AUTOINCREMENT,
    name  TEXT NOT NULL,
    brand TEXT NOT NULL,
    price REAL NOT NULL
)
""")

# 清空旧数据，方便反复练习
cur.execute("DELETE FROM products")
conn.commit()          # 删也是"动数据"，同样要盖章

# 增：批量插入 5 个商品（executemany 一口气）
products = [
    ("机械键盘",   "樱桃",     399.0),
    ("无线鼠标",   "罗技",     129.0),
    ("显示器",     "戴尔",    1299.0),
    ("耳机",       "索尼",     899.0),
    ("人体工学椅", "赫曼米勒", 5299.0),
]
cur.executemany(
    "INSERT INTO products (name, brand, price) VALUES (?, ?, ?)",
    products,
)
conn.commit()          # 不 commit，这 5 行全丢

# 查：全部商品，价格从高到低
cur.execute("SELECT name, brand, price FROM products ORDER BY price DESC")
for name, brand, price in cur.fetchall():
    print(f"{name}（{brand}）：{price:.0f} 元")

# 改：机械键盘降价 100
cur.execute("UPDATE products SET price = price - 100 WHERE name = ?", ("机械键盘",))
conn.commit()          # 改也要盖章

# 删：下架耳机
cur.execute("DELETE FROM products WHERE name = ?", ("耳机",))
conn.commit()          # 删也要盖章

# 验证：看最终结果
cur.execute("SELECT COUNT(*) FROM products")
print("还剩", cur.fetchone()[0], "个商品")

conn.close()

运行后输出：

人体工学椅（赫曼米勒）：5299 元
显示器（戴尔）：1299 元
耳机（索尼）：899 元
机械键盘（樱桃）：399 元
无线鼠标（罗技）：129 元
还剩 4 个商品

注意一个小细节：三个 ? 就是占位符，值放在 execute 的第二个参数（元组）里。为什么非要这么写、不能图省事用 f-string 把值拼进 SQL？这是今天最重要的伏笔，第五节揭晓。

四、两个”忘了就出事”的家伙：commit 和 close

忘 commit：程序不报错，数据却不翼而飞。这是新手第一大灵异事件——”我明明 INSERT 了，怎么表里是空的？”做个实验就明白：

# ghost.py — 数据"鬼消失"实验
import sqlite3

conn = sqlite3.connect("ghost.db")
cur = conn.cursor()
cur.execute("CREATE TABLE IF NOT EXISTS log (msg TEXT)")
cur.execute("INSERT INTO log VALUES ('这条没盖章')")
# 忘了 conn.commit()
conn.close()

# 重新打开验证
conn = sqlite3.connect("ghost.db")
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM log")
print("表里有几行：", cur.fetchone()[0])   # 输出：表里有几行： 0

程序全程不报错，行就是没了。原因回到档案馆的比方：没盖章的指令只是”口头说了一嘴”，连接一关，便签直接作废。疫苗一句话：凡是有增、删、改的代码，每次动完数据紧跟一行 conn.commit()；拿不准就多盖一次章——commit 本身没有副作用，数据不动的时候盖章等于没盖。

忘 close：专线一直占着。小脚本跑完进程退出，系统会替你收尾；但如果是长期运行的程序（比如每天跑一次的价格监控），连接越积越多，最后”too many open files”。稳妥写法是把收尾交给 try/finally，或者干脆封装成函数：

def add_product(name, brand, price):
    """加一个商品，连接自动收尾"""
    conn = sqlite3.connect("shop.db")
    try:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO products (name, brand, price) VALUES (?, ?, ?)",
            (name, brand, price),
        )
        conn.commit()
    finally:
        conn.close()      # 不管出不出错，电话一定挂

顺手打一个认知疫苗：Python 的 with 语句用在连接上，是 with sqlite3.connect(...) as conn:——它管的是事务（块内出错自动回滚、正常走完自动 commit），但它不替你 close 连接。所以别以为套个 with 就万事大吉，close 还是得自己安排（finally 或显式调用）。

五、重头戏：SQL 注入——字符串拼出来的”万能钥匙”

假设你要给价格监控工具加个”按商品名查价格”的功能，图省事这么写：

# 危险代码！仅作反面教材，别照抄
def find_price(name):
    cur.execute(f"SELECT price FROM products WHERE name = '{name}'")
    return cur.fetchone()

正常查 find_price("机械键盘") 没问题。但如果 name 来自用户输入（网页表单、配置文件、爬来的数据），别人输入这么一串会发生什么：

' OR '1'='1

拼进 SQL 就变成了：

SELECT price FROM products WHERE name = '' OR '1'='1'

'1'='1' 永远为真——WHERE 形同虚设，整张表的数据被全部拖走。这就是大名鼎鼎的SQL 注入：把”数据”伪装成”指令”，从你亲手开的口子里混进档案馆改档案、删档案、偷档案。更狠的输入还能跟一句 '; DROP TABLE products; --，连桌子都给你掀了。程序不报错、不留痕迹，中招的往往还是”自己人写的内部工具”。

解药你其实早就会了——第三节埋的伏笔：参数化查询（占位符）：

# 安全版本
def find_price(name):
    cur.execute("SELECT price FROM products WHERE name = ?", (name,))
    row = cur.fetchone()
    return row[0] if row else None

传入 ' OR '1'='1 时，数据库收到的指令是”查 name 等于这串字符的商品”——它被当成一整个数据值老老实实去匹配，匹配不上就返回空，指令部分毫发无伤。占位符的精髓：数据永远是数据，指令永远是指令，两者永不越界。

两条补充疫苗：① 占位符在 SQLite 里是 ?，换别的数据库（MySQL、PostgreSQL）写法可能不同，让 AI 写的时候说清用的是哪个数据库；② 表名、列名不能用占位符（它们是指令的一部分），如果确实要动态拼表名，只能自己做白名单校验——只允许预定义的那几个名字。

六、三个必踩的坑（先打疫苗）

坑1：增删改之后忘 commit，程序”正常”跑完、数据没存上。和所有静默失败一样，它不报错，等你第二天打开数据库才发现是空的。疫苗一句话：动数据的每一段，收尾标配两行——conn.commit() 然后 conn.close()；写完自查一遍代码里 INSERT/UPDATE/DELETE 的次数和 commit 的次数，动了三次就该有三次盖章（或一次统一盖章，但必须有）。

坑2：f-string 拼 SQL，看着方便，实际是给注入开门。哪怕数据来自”自己爬的””自己程序写的”，也一律用 ? 占位符——今天安全不代表明天安全（哪天这个函数被复用到带用户输入的地方呢）。疫苗一句话：SQL 语句是常量，值走元组参数，养成肌肉记忆；拿到 AI 生成的代码，看见 f-string / format / + 拼进 execute 的 SQL，一律打回重写。

坑3：把查询结果当”永远有数据”。cur.fetchone() 查不到时返回的是 None，直接 row[0] 当场 TypeError: 'NoneType' object is not subscriptable；fetchall() 返回的是列表的列表 [(399.0,), ...]，别当成平铺的 [399.0, ...] 来用。疫苗一句话：fetchone 的结果先判空再取值，fetchall 的每一行再取一次下标。顺带一提，游标用完一次查询，想再查一轮直接重新 execute 就行，上一轮没取完的结果会被冲掉——别指望”存着慢慢取”。

七、跟AI协作的正确姿势（数据库代码特供版）

  1. 给 AI 立规矩，比验收更省事。提需求时加一句：”用 sqlite3 标准库，SQL 一律参数化查询，禁止字符串拼接；动数据后必须 commit，连接用 try/finally 关闭。”规矩说在前面，AI 产出的代码大概率直接就是安全骨架。
  2. 验收三查，一分钟搞定。拿到数据库代码只查三样：① execute 里的 SQL 是不是常量 + ? 占位符；② 每段增删改后面有没有 commit；③ conn.close() 有没有兜底（finally 或 with 的正确用法）。三样都过，再跑功能。
  3. 报障带上”三层现场”。“数据没存上”是差需求；”INSERT 后打印了执行成功、程序正常退出、重开数据库 COUNT=0，代码如下”是好需求——现象、验证步骤、代码三层贴齐，AI 一眼锁定”没 commit”。
  4. 让 AI 帮你写”体检脚本”。不确定有没有漏 commit？直接让 AI 写个小脚本：跑一遍你的程序，再独立打开数据库数行数对不对。数据不对时让 AI 先查”流程四件套”，再怀疑 SQL 本身。

八、今日小练习

练习1：把 JOIN 包成函数。沿用上一篇的 shop.db（products + prices 两张表），写一个函数 brand_prices(category)：输入品类（如”外设”），返回该品类所有价格记录的列表（每条含商品名、品牌、价格，价格从高到低）。要求全程占位符 + commit/close 兜底。

提示：JOIN + WHERE 组合过滤品类；函数里新建连接、try/finally 关闭；fetchall 返回后可以整理成字典列表再返回，用起来更顺手。

练习2：亲手做一次注入实验（安全环境）。先写一个故意用 f-string 拼 SQL 的查询函数，在本地练习库里传入 ' OR '1'='1，观察”整表被拖走”的现象；然后改成参数化版本，再传同样的输入，对比结果。做完把危险版函数删掉或加注释封印。

提示：实验就在自己的练习库里做，别拿别人的系统练手；对比两次的返回行数，一眼看出差别；想想”如果这是登录功能，会出什么事”。

练习3（坑点自考）：先诊断，再对答案。下面每段都有毛病，不看文章说出病在哪、怎么改：①程序里 cur.execute("INSERT INTO products ...") 之后直接 conn.close()，重开数据库表是空的；②查询写成 cur.execute(f"SELECT * FROM products WHERE brand = '{brand}'")，拿用户输入当 brand；③price = cur.fetchone()[0] 在空表上直接崩，报 TypeError: 'NoneType' object is not subscriptable。

提示：①档案馆的规矩，口头说不算数；②数据想混进指令队伍，得有个”隔离墙”关键词；③取值之前先看看手里拿到的是不是 None。

下篇预告

下一篇（第56篇）：数据库+Flask — 让网页替你操作数据库。数据终于住进了正经数据库，可总不能让家人朋友也来敲 Python 命令行吧？下一篇把数据库接上 Flask，做个能用浏览器打开的小网页：页面上点一点，数据就增删改查——你的”正经软件”第一次有了脸面。sqlite3 四件套 + 参数化查询今天都备齐了，正好全套用上。