开始前,会这些就够了
- 知道父表、子表和主键的基本含义。
- 能读懂 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。若失败,保留错误检查运行库能力与事务状态,不要删掉断言使演示继续。
展开收起代码 · 小实验
长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。
"""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 执行未报错等同于已生效。
- 认为开关写进了数据库文件,后续连接会自动沿用。
- 看到审计为空就推断全部业务正确,或直接批量删除真实孤儿数据。
自己试一小步
如果一份教学库有两类子记录,怎样区分某张表审计干净与整个库没有外键违规?
- 在脚本副本增加另一张子表与不同编号,先画出引用关系。
- 自行安排两个连接与事务阶段,写出每阶段开关和预计错误,不沿用本文顺序。
- 分别运行限定表名和未限定表名的审计,保留四列完整结果。
- 为每条虚构违规说明修复依据,修复后同时核对数据总量和关联。
做到这些,就算完成
- 预测与实际结果分开保存。
- 能指出命令在哪条连接和哪个事务状态下执行。
- 没有把某张表的结果扩大成全部数据库结论。
- 没有访问真实业务库,也没有把顺序测试当成并发证明。
查原始资料
想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。
- SQLite:启用外键支持 (新标签页)↗
核读第 1、2 节:关联约束、构建支持、每连接启用,以及活动事务内改变开关无效。
- SQLite:foreign_keys 与 foreign_key_check (新标签页)↗
核读
foreign_key_check、foreign_key_list、foreign_keys三项:检查范围、四列含义与事务中的 no-op。 - Python 3.14:sqlite3 事务控制与关闭连接 (新标签页)↗
核读
isolation_level、LEGACY_TRANSACTION_CONTROL、in_transaction与连接上下文管理器说明,支撑显式事务和关闭责任。