Python SQLite 外键不生效：连接开关、整批回滚与历史孤儿记录检查

SQLite 表里已经写了 REFERENCES，批处理却仍能插入不存在的任务 ID。问题可能不在 SQL 拼写，而在当前连接没有启用外键检查。本文用一个不接触真实数据库的内存实验，对比开关前后的同一批输入，再验证整批回滚和历史孤儿记录检查，适合初级到中级读者。

1. 方法：建约束、开连接检查、再做历史审计

外键约束描述表之间的关系；运行时还要确认连接是否执行它。不要假设服务器上的 SQLite 构建默认开启检查：连接后主动设置 PRAGMA foreign_keys = ON，紧接着查询返回值，要求为 1。连接池或第二个进程建立的连接也需要这一步。SQLite 官方启用说明明确说明了连接级开关及事务内不能切换的限制。

这里的父表 jobs 存任务，子表 results 存结果。results.job_id 必须指向已有任务；同时加 NOT NULL，避免空值绕过这一必填关系。示例不设置级联删除：有子记录时，删除父任务应失败。

这不是备份替代品，也不是对现有库的自动修复工具。操作真实数据前，先做 SQLite 备份和恢复验证；本页所有数据库都只存在于内存中。

2. 准备：只使用 Python 标准库

需要 Python 3.11 或更新版本和可用的 sqlite3 模块，无需 pip 包或 SQLite 命令行。示例在 Linux、Python 3.11.4 下测试。按顺序执行，后续命令留在同一个新目录：

python3 --version
python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
mkdir sqlite-fk-demo
cd sqlite-fk-demo

将下一节保存为 fk_demo.py，也可下载完整演示脚本。它只接受 --enforce on 或 --enforce off，默认 on；off 仅用于故障对照。

3. 实现：在 BEGIN 之前设置并核对开关

"""Compare foreign-key enforcement in a disposable in-memory database."""
import argparse
import json
import sqlite3
import sys


def open_db(enforce=True):
    # No implicit BEGIN: all write transactions below are explicit.
    con = sqlite3.connect(":memory:", isolation_level=None)
    try:
        con.execute("PRAGMA foreign_keys = " + ("ON" if enforce else "OFF"))
        expected = (1 if enforce else 0,)
        if con.execute("PRAGMA foreign_keys").fetchone() != expected:
            raise RuntimeError("foreign-key setting was not applied")
        con.execute("CREATE TABLE jobs (id INTEGER PRIMARY KEY)")
        con.execute("""CREATE TABLE results (
            id INTEGER PRIMARY KEY,
            job_id INTEGER NOT NULL REFERENCES jobs(id)
        )""")
        con.execute("CREATE INDEX results_job_id ON results(job_id)")
        con.execute("INSERT INTO jobs VALUES (1)")
        return con
    except Exception:
        con.close()
        raise


def run_demo(enforce=True):
    con = open_db(enforce)
    error = None
    try:
        try:
            con.execute("BEGIN")
            con.execute("INSERT INTO results VALUES (?, ?)", (1, 1))
            con.execute("INSERT INTO results VALUES (?, ?)", (2, 999))
            con.execute("COMMIT")
        except sqlite3.IntegrityError as exc:
            con.execute("ROLLBACK")
            error = str(exc)
        report = {
            "accepted": error is None,
            "foreign_keys": con.execute("PRAGMA foreign_keys").fetchone()[0],
            "result_rows": con.execute("SELECT COUNT(*) FROM results").fetchone()[0],
            "violations": con.execute("PRAGMA foreign_key_check").fetchall(),
        }
        return report, error
    finally:
        con.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--enforce", choices=("on", "off"), default="on")
    args = parser.parse_args()
    report, error = run_demo(args.enforce == "on")
    print(json.dumps(report, sort_keys=True))
    if error:
        print("ERROR: " + error, file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

isolation_level=None 关闭 Python 的隐式事务开启行为；写入批次明确执行 BEGIN、COMMIT 或 ROLLBACK。参见 Python sqlite3 事务控制。这里没有使用 Python 3.12 才增加的 autocommit 参数；不要混入一个总是保持事务打开的连接策略。

先建立任务 1，再尝试同一批写入 (1, 1) 和 (2, 999)。第二条记录引用不存在的任务：开启检查时，异常处理显式回滚整个批次，因此第一条有效结果也不会留下。单条语句失败不等于前面所有写入自动撤销。

results_job_id 索引方便按父键查找子记录；它不是启用外键的开关。SQL 值通过 ? 绑定，不用字符串拼接；PRAGMA 中拼接的仅是代码内固定的 ON、OFF。

4. 验证：同一批输入，比较行数与退出码

python3 fk_demo.py --help
python3 fk_demo.py --enforce off
if python3 fk_demo.py --enforce on; then
  printf 'UNEXPECTED: invalid batch accepted\n' >&2
  exit 1
else
  rc=$?
  printf 'protected exit=%s\n' "$rc"
  test "$rc" -eq 1
fi

关闭检查时，标准输出如下，退出码为 0，标准错误为空：

{"accepted": true, "foreign_keys": 0, "result_rows": 2, "violations": [["results", 2, "jobs", 0]]}

开启检查时，标准输出如下；标准错误是 ERROR: FOREIGN KEY constraint failed，脚本退出 1，shell 随后打印 protected exit=1：

{"accepted": false, "foreign_keys": 1, "result_rows": 0, "violations": []}

result_rows: 0 是整批回滚的结果；accepted: false 仍是失败报告，不能因为 stdout 有 JSON 就当作成功。自动化必须同时检查退出码。错误的 --enforce 值由 argparse 返回 2，帮助返回 0。

可下载回归测试，和脚本放在同一目录后运行：

python3 test_fk_demo.py

测试会检查退出码、结果行数、stderr、历史违规、事务内开关无效以及拒绝删除被引用父行。它不连接生产数据库，也不会生成数据库文件。

5. 检查历史违规：开启不等于修复

下面另存为 legacy_check.py，在演示目录执行 python3 legacy_check.py：

from fk_demo import open_db

con = open_db(False)  # Reproduce a legacy writer only; not a production setting.
try:
    con.execute("INSERT INTO results VALUES (7, 999)")
    con.execute("PRAGMA foreign_keys = ON")
    print("enabled:", con.execute("PRAGMA foreign_keys").fetchone()[0])
    print("violations:", con.execute("PRAGMA foreign_key_check").fetchall())
finally:
    con.close()

输出：

enabled: 1
violations: [('results', 7, 'jobs', 0)]

PRAGMA 官方说明定义了每行的四项含义：子表、问题行的 rowid、父表、外键编号。开启检查不会删除或修复已经存在的孤儿记录；这里的 7 是普通 rowid 表中出问题的行。

迁移旧库时，先在稳定备份或停止写入的副本上运行 PRAGMA foreign_key_check。非空结果应阻止验收，然后按业务依据补父记录、修正引用或隔离坏数据，不要直接批量删除。没有声明外键的表不会因这条命令获得关系校验；空结果也不是完整的业务验证或磁盘完整性保证。

6. 排障：先区分约束失败与锁冲突

现象	直接处理
设置 ON 后仍查询到 `0`	检查 `con.in_transaction`；先明确提交或回滚当前业务事务，再在事务外设置。不要为了切换开关而盲目提交未验收数据。
`PRAGMA foreign_keys` 返回空行	检查 SQLite 是否具备外键支持；保留连接预检，让任务失败，而不是带着未知状态继续。
`FOREIGN KEY constraint failed`	核对父 ID 是否已存在、导入顺序是否先父后子，并回滚失败批次；不要用 OFF 作为修复。
`foreign key mismatch`	检查父键是否是主键或满足外键要求的唯一键，并核对列数和排序规则；改表前先备份。
重连之后又能写入坏记录	将设置与查询集中在连接工厂；每个写入入口都必须调用。
`database is locked`	这是写入竞争，不是缺少父键；按短事务与有限等待排障处理。

如果批处理需要断点续跑，把结果和进度的数据库修改放进同一事务。外键只能约束数据库内的关系，不保证外部文件、HTTP 请求或整个工作流恰好执行一次。

总结

声明外键之后，还要为每个连接在事务外启用并核对检查状态。用同一批合法与非法输入验证拒绝和整批回滚，再单独审计历史违规。保留备份和业务修复依据，不把关闭检查当作修复手段。