qinghuan
← 回到知识探索

后端现场 / 动手实验

外键写在表里,为什么还是插进了不存在的编号?

让两个真实 SQLite 连接访问同一份临时数据库,对照约束定义、当前连接的执行开关,以及已有坏记录的单独审计。

展开本文目录
  1. 下载完整练习包
  2. 先把定义、开关和现有数据分成三个问题
  3. 同一份文件,两条连接可以执行不同检查
  4. 事务已经开始,再打开开关可能没有任何效果
  5. 成功开启后,旧孤儿不会自动消失
  6. 修复是业务决定,检查不会替你选择答案
  7. 运行五阶段实验,把每一条连接写进记录
  8. 把连接状态核验放到真正的写入入口
  9. 记住这三点
  10. 别把例子用过头
  11. 自己试一小步
  12. 查原始资料
开始前,会这些就够了
  • 知道父表、子表和主键的基本含义。
  • 能读懂 INSERT、DELETE、BEGIN 与 ROLLBACK。
  • Python 3.10+,其 SQLite 构建支持外键检查。

一张借阅记录写着读者编号九百九十九,但读者表里根本没有这个人。你查看建表语句,REFERENCES 也在。为什么数据库仍然收下了记录?在 SQLite 中,需要把“表声明了关联”与“执行这次写入的连接启用了检查”分别核对。

本篇打开两个连接,让它们面对同一份临时文件,却使用不同的外键开关。你会得到五个阶段的状态表和一份违规记录清单,看见写入约束、运行状态与历史审计怎样回答不同问题。

把例子带到自己的电脑上

下载完整练习包

需要 Python 3.10 或更新版本,无第三方依赖;制作时实测 Python 3.14.0。使用标准库,不需要安装第三方包。下载 ZIP 后选择“全部提取”,进入能看到 README.md 的文件夹。

下载 ZIP · 5 个文件
包里有什么
  • foreign_key_connections.py:完整示例
  • README.md:环境、范围与独立练习
  • observed.stdout.txt:制作时的实际输出
  • verification.json:源码与产物验证记录
  • SHA256SUMS.txt:文件内容摘要

在这个文件夹打开 PowerShell,逐行运行

python --version
python -B foreign_key_connections.py

解压到独立练习文件夹,在含源文件的目录打开终端运行。程序会在当前目录写入教学报告;先预测结果,再查看当次输出。真实临时 SQLite 文件、两个连接及受控事务顺序;不代表其他数据库、线上迁移或并发压力测试。

找不到命令或文件时,先按 README 检查环境和所在目录。文中标出的阅读时间只估算文章篇幅,安装环境和动手练习可按自己的节奏进行。

先把定义、开关和现有数据分成三个问题

实验只有 parent 与 child 两张表,父表先放入编号一。子表的 parent_id 必填,并声明引用父表主键。这样我们集中观察“填写的编号没有对应父记录”,不混入未填写值、级联删除或延迟约束的其他规则。

建表语句描述允许的关联;PRAGMA foreign_keys 返回这个连接是否执行外键检查;PRAGMA foreign_key_check 列出现有数据中的外键违规。看见第一项,不能代替第二项;第二项已经开启,也不能据此省略对既有数据的第三项检查。

脚本会读取两个新连接最初的值,但不会据此设计实验。随后明确把第一条连接设为开启、第二条设为关闭。这样换到默认值不同的构建时,实验仍然比较同样两种状态;若运行库根本不支持检查,则主动失败。

同一份文件,两条连接可以执行不同检查

先通过第一条连接插入子记录十,引用不存在的九百九十九,程序收到外键完整性错误。接着用第二条连接插入子记录二十,引用同一个不存在编号,这次写入成功。两条连接的开关仍分别为一和零,没有随着文件内容共享。

随后第一条连接运行审计,也看到了第二条连接留下的孤儿记录。“孤儿”在这里就是找不到被引用父记录的子行。第一条连接自己设置正确,并不能阻止其他未检查的写入路径破坏这份共同数据。

这也解释了连接池为什么需要关注每条新建连接的初始化。不要只在启动时借出一条连接执行命令,就推断后面所有连接已采用相同设置。本篇没有接入连接池,只用两条真实连接把这个范围差异显露出来。

事务已经开始,再打开开关可能没有任何效果

第二条连接当前仍关闭检查。程序明确 BEGIN 后执行 PRAGMA foreign_keys = ON,命令没有报错,但重新读取仍是零。它继续写入子记录二十一,引用另一个不存在编号;此时在这条事务内审计,可以看到两条违规。

这不是代码忘了提交开关,也不是 SQL 返回成功就代表状态变化。在 SQLite 中,活动事务内尝试改变这个开关不起作用。本例紧跟命令查询实际值,并确认事务仍活动,直接留下“请求了什么”和“最终用了什么”的对照。

接着 ROLLBACK 撤销本次事务里的子记录二十一,原来已经写入的二十仍在。实验通过显式事务控制固定这个顺序;Python 3.12 及以上也明确选择对应旧式控制模式,避免自动开启事务改变开关执行的位置。

成功开启后,旧孤儿不会自动消失

事务结束后,第二条连接再次开启检查,这次读回一。它再尝试插入子记录二十二时被拒绝,但前面已留下的二十仍能查到。开启检查约束的是后续相关写入,不等于执行了一次修复程序。

审计输出的四列分别是子表名、子行的 rowid、父表名和外键编号。本文 child 的 INTEGER PRIMARY KEY 对应 rowid,所以结果里的二十能直接定位该行;不要把这种对应关系推广到所有表,WITHOUT ROWID 表在这一列的规则也不同。

foreign_key_check 针对外键关联,不是完整业务体检。它不会替你证明金额合理、身份有权限或名称没有写错。带表名的形式也只检查该表定义的外键;本程序使用未限定表名的形式检查当前主数据库,并没有附加其他数据库。

修复是业务决定,检查不会替你选择答案

教学数据明确约定二十是一条错误录入,因此脚本按固定编号删除它,再加入合法子记录三十,引用父记录一。这个修复只针对我们自己创建的虚构行。真实资料里可能应该补建父记录、修正引用或转交人工处理,不能看到孤儿就批量删除。

最后通过第一条已开启检查的连接尝试删除父记录一,操作被拒绝,因为子记录三十还在引用它。这一步把检查方向补全:不仅新子行可能出错,删除或修改被引用的数据也会影响关联。本文没有设置级联动作,不能从这次拒绝推断所有外键都拒绝父行删除。

最终审计为空,子表完整列表只有三十指向一。我们同时检查违规清单与业务记录,避免以“违规数为零”掩盖误删全部数据的另一种错误。返回码、状态开关和最终记录应当能够相互核对。

运行五阶段实验,把每一条连接写进记录

将完整代码保存为 foreign_key_connections.py 后执行。它在系统临时目录创建数据库,依次使用两条连接,结束后关闭并清理自己的目录;不接受用户数据库路径。脚本旁的 foreign-key-output 会保留 report.json 与 stages.csv,再次运行会更新这些教学产物。

本次 Python 3.14.0、SQLite 3.50.4 实跑通过,默认观察值是零、零,但程序没有依赖它。五阶段的违规数依次为零、一、二、一、零;第三阶段包含尚未提交的当前事务数据,应连同 stage 阅读,不能当成五次最终提交结果。

先预测哪个动作会报错、哪个不报错却没有效果,再核对 trace 和 old_violations_after_enabling。末行 PASS 之外,temporary_directory_removed 也应为 true。若失败,保留错误检查运行库能力与事务状态,不要删掉断言使演示继续。

展开收起代码 · 小实验python · 94 行

长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。

小实验 · python
"""Connection-local enforcement and retrospective audit, on a temporary SQLite file."""
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
import csv
import json
import sqlite3
import sys


def connect(path):
    options = {"isolation_level": None}
    if sys.version_info >= (3, 12):
        options["autocommit"] = sqlite3.LEGACY_TRANSACTION_CONTROL
    return sqlite3.connect(path, **options)


def enabled(db):
    row = db.execute("PRAGMA foreign_keys").fetchone()
    assert row is not None, "SQLite build has no foreign-key enforcement support"
    return row[0]


def audit(db):
    return [list(row) for row in db.execute("PRAGMA foreign_key_check")]


def rejected(db, statement, values):
    try:
        db.execute(statement, values)
    except sqlite3.IntegrityError as error:
        assert "FOREIGN KEY" in str(error), error
        return True
    raise AssertionError("expected a foreign-key violation")


def main():
    trace = []
    with TemporaryDirectory(prefix="fk-lesson-") as directory:
        root = Path(directory)
        with closing(connect(root / "demo.db")) as first, closing(connect(root / "demo.db")) as second:
            first.execute("CREATE TABLE parent(id INTEGER PRIMARY KEY)")
            first.execute("CREATE TABLE child(id INTEGER PRIMARY KEY, parent_id INTEGER NOT NULL REFERENCES parent(id))")
            first.execute("INSERT INTO parent VALUES (1)")
            defaults = [enabled(first), enabled(second)]
            # Deliberately choose both states; do not depend on a build's defaults.
            first.execute("PRAGMA foreign_keys = ON")
            second.execute("PRAGMA foreign_keys = OFF")
            assert (enabled(first), enabled(second)) == (1, 0)
            trace.append({"stage": "two_connections", "first": 1, "second": 0, "violations": 0})
            assert rejected(first, "INSERT INTO child VALUES (?, ?)", (10, 999))
            second.execute("INSERT INTO child VALUES (?, ?)", (20, 999))
            assert audit(first) == [["child", 20, "parent", 0]]
            trace.append({"stage": "second_inserted_orphan", "first": enabled(first), "second": enabled(second), "violations": len(audit(first))})
            second.execute("BEGIN")
            second.execute("PRAGMA foreign_keys = ON")
            assert enabled(second) == 0 and second.in_transaction
            second.execute("INSERT INTO child VALUES (?, ?)", (21, 998))
            assert len(audit(second)) == 2
            trace.append({"stage": "enable_inside_transaction_ignored", "first": enabled(first), "second": enabled(second), "violations": len(audit(second))})
            second.execute("ROLLBACK")
            second.execute("PRAGMA foreign_keys = ON")
            assert enabled(second) == 1 and len(audit(second)) == 1
            assert rejected(second, "INSERT INTO child VALUES (?, ?)", (22, 997))
            old_violations = audit(second)
            trace.append({"stage": "enabled_but_old_orphan_remains", "first": enabled(first), "second": enabled(second), "violations": len(old_violations)})
            # Explicit fixture repair chosen for this made-up row only.
            second.execute("DELETE FROM child WHERE id = ?", (20,))
            second.execute("INSERT INTO child VALUES (?, ?)", (30, 1))
            assert rejected(first, "DELETE FROM parent WHERE id = ?", (1,))
            assert audit(first) == []
            final_rows = [list(r) for r in first.execute("SELECT id, parent_id FROM child ORDER BY id")]
            assert final_rows == [[30, 1]]
            trace.append({"stage": "fixture_repaired_and_parent_delete_blocked", "first": enabled(first), "second": enabled(second), "violations": 0})
    assert not root.exists()
    report = {"python": sys.version.split()[0], "sqlite": sqlite3.sqlite_version,
              "observed_defaults_not_assumed": defaults, "trace": trace,
              "audit_columns": ["child_table", "child_rowid", "parent_table", "foreign_key_id"],
              "old_violations_after_enabling": old_violations,
              "final_child_rows": final_rows, "temporary_directory_removed": True,
              "scope": "two real connections used sequentially; no concurrent contention or production database"}
    out = Path(__file__).resolve().parent / "foreign-key-output"
    out.mkdir(exist_ok=True)
    (out / "report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    with (out / "stages.csv").open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=["stage", "first", "second", "violations"])
        writer.writeheader()
        writer.writerows(trace)
    print(json.dumps(report, indent=2))
    print("PASS: per-connection flags, transaction no-op, historical audit, restricted delete and cleanup")


if __name__ == "__main__":
    main()

怎样运行:python -B foreign_key_connections.py(Python 3.10+,仅标准库;在解压后的脚本目录执行。) 2026-10-15 已用 Python 3.14.0 执行这份代码并通过自带检查;下方是当次输出。真实临时 SQLite 文件、两个连接及受控事务顺序;不代表其他数据库、线上迁移或并发压力测试。

运行后对照结果
{
  "python": "3.14.0",
  "sqlite": "3.50.4",
  "observed_defaults_not_assumed": [
    0,
    0
  ],
  "trace": [
    {
      "stage": "two_connections",
      "first": 1,
      "second": 0,
      "violations": 0
    },
    {
      "stage": "second_inserted_orphan",
      "first": 1,
      "second": 0,
      "violations": 1
    },
    {
      "stage": "enable_inside_transaction_ignored",
      "first": 1,
      "second": 0,
      "violations": 2
    },
    {
      "stage": "enabled_but_old_orphan_remains",
      "first": 1,
      "second": 1,
      "violations": 1
    },
    {
      "stage": "fixture_repaired_and_parent_delete_blocked",
      "first": 1,
      "second": 1,
      "violations": 0
    }
  ],
  "audit_columns": [
    "child_table",
    "child_rowid",
    "parent_table",
    "foreign_key_id"
  ],
  "old_violations_after_enabling": [
    [
      "child",
      20,
      "parent",
      0
    ]
  ],
  "final_child_rows": [
    [
      30,
      1
    ]
  ],
  "temporary_directory_removed": true,
  "scope": "two real connections used sequentially; no concurrent contention or production database"
}
PASS: per-connection flags, transaction no-op, historical audit, restricted delete and cleanup

把连接状态核验放到真正的写入入口

带回应用时,先列出会创建连接的入口:正常请求、后台任务、导入脚本和维护程序。为每条需要检查的连接规定初始化位置,并读取实际状态确认。只检查配置文件,可能漏掉初始化发生得太晚,或者某个旁路完全没有执行的情况。

这次确实用了两个 SQLite 连接和一个文件,但操作按明确顺序执行,没有竞争、锁等待或故障恢复试验。审计为空只描述那个检查时刻的样本;如果仍允许其他连接关闭检查后写入,结果之后仍可能变化。

本篇也不代表 MySQL、PostgreSQL 或任意 ORM 都采用同样开关规则。可迁移的排查方法是:把声明、当前执行状态、写入结果和已有数据分别留下证据;具体命令与作用域仍按正在使用的数据库核对。

记住这三点

  • SQLite 外键执行开关属于连接,表定义与一个连接的正确设置不能覆盖所有写入路径。
  • 活动事务内切换 foreign_keys 没有效果,应读取实际值确认。
  • 启用检查不会自动修复旧违规;foreign_key_check 也只回答外键关联问题。

别把例子用过头

  • 把一次 PRAGMA 执行未报错等同于已生效。
  • 认为开关写进了数据库文件,后续连接会自动沿用。
  • 看到审计为空就推断全部业务正确,或直接批量删除真实孤儿数据。

自己试一小步

如果一份教学库有两类子记录,怎样区分某张表审计干净与整个库没有外键违规?

  1. 在脚本副本增加另一张子表与不同编号,先画出引用关系。
  2. 自行安排两个连接与事务阶段,写出每阶段开关和预计错误,不沿用本文顺序。
  3. 分别运行限定表名和未限定表名的审计,保留四列完整结果。
  4. 为每条虚构违规说明修复依据,修复后同时核对数据总量和关联。

做到这些,就算完成

  • 预测与实际结果分开保存。
  • 能指出命令在哪条连接和哪个事务状态下执行。
  • 没有把某张表的结果扩大成全部数据库结论。
  • 没有访问真实业务库,也没有把顺序测试当成并发证明。

查原始资料

想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。

  • SQLite:启用外键支持 (新标签页)↗SQLite 官方文档 · 英文

    核读第 1、2 节:关联约束、构建支持、每连接启用,以及活动事务内改变开关无效。

    2026-10-15 实际核读所列范围;未把默认值当成程序前提;未测试级联或延迟约束。

  • SQLite:foreign_keys 与 foreign_key_check (新标签页)↗SQLite 官方文档 · 英文

    核读 foreign_key_check、foreign_key_list、foreign_keys 三项:检查范围、四列含义与事务中的 no-op。

    2026-10-15 实际核读所列范围;实际实验只有主数据库的普通 rowid 表,不含 ATTACH 或 WITHOUT ROWID。

  • Python 3.14:sqlite3 事务控制与关闭连接 (新标签页)↗Python 官方文档 · 英文

    核读 isolation_level、LEGACY_TRANSACTION_CONTROL、in_transaction 与连接上下文管理器说明,支撑显式事务和关闭责任。

    2026-10-15 实际核读所列范围;实际执行 3.14.0;较旧版本分支未在本轮逐版本运行。