SQLite 已声明外键却写入孤儿数据：连接初始化与存量校验实践

订单明细表已经写了 REFERENCES orders(id)，导入程序却能插入不存在的订单号。更容易误判的情况是：初始化脚本设置过 PRAGMA foreign_keys = ON，接口仍然产生孤儿明细。

这时先确认写入连接的状态。SQLite 的外键执行开关属于连接；迁移连接开启，不代表接口、后台任务和临时导入脚本的连接也开启。下面用订单与明细两个表复现问题，再把检查放进连接创建入口。适用于 Python 标准库 sqlite3 和本地 SQLite 数据库，不涉及 PostgreSQL、MySQL 的配置。

1. 在故障连接上取证

不要为了检查而另开一个客户端连接。应通过真正执行故障写入的连接执行：

PRAGMA database_list;
PRAGMA foreign_keys;
PRAGMA foreign_key_list(order_lines);
PRAGMA foreign_key_check;

依次确认数据库文件路径、当前执行开关、表的外键声明和存量违规记录。foreign_keys 返回 1 才说明当前连接开启；没有返回行时应检查 SQLite 版本与编译支持，不应当作开启。

foreign_key_check 每条结果包含子表名、子表 rowid、父表名、外键序号；最后一列可与 foreign_key_list 的 id 对应。没有结果表示这次检查没有发现外键违规；它不是全库业务正确性检查。WITHOUT ROWID 表的违规行不会提供普通 rowid。字段定义见官方 PRAGMA 文档。

先排除连错文件，再判断开关问题。相对路径、工作目录变化和同名测试库，都可能让取证与故障发生在不同文件。

2. 复现两个容易漏掉的入口

下面只在 SQLite 临时内存库执行：

PRAGMA foreign_keys = OFF;
CREATE TABLE orders (id INTEGER PRIMARY KEY);
CREATE TABLE order_lines (
    id INTEGER PRIMARY KEY,
    order_id INTEGER NOT NULL REFERENCES orders(id)
);
INSERT INTO order_lines VALUES (1, 999);
PRAGMA foreign_key_check;

BEGIN;
PRAGMA foreign_keys = ON;
PRAGMA foreign_keys;
ROLLBACK;

PRAGMA foreign_keys = ON;
PRAGMA foreign_keys;
PRAGMA foreign_key_check;

第一条明细能够插入，违规检查返回 order_lines|1|orders|0。事务中的开关读取仍为 0；结束事务后设置才变为 1，但已存在的违规行仍然能被检查出来。

复现主动关闭开关，是为了让结果不依赖发行版的默认设置。真实程序应显式开启并读取确认。另一个关键点是设置时机：在已经开始的事务中执行开关设置不会报错，却不会生效。SQLite 官方外键说明明确要求逐连接启用，并说明了事务中的限制。

3. 在连接创建后、事务开始前完成初始化

将下面代码保存为 test_sqlite_foreign_keys.py。它只使用标准库；后面测试代码追加到同一文件。

import sqlite3
import unittest
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory


SCHEMA = """
CREATE TABLE orders (id INTEGER PRIMARY KEY);
CREATE TABLE order_lines (
    id INTEGER PRIMARY KEY,
    order_id INTEGER NOT NULL REFERENCES orders(id)
);
CREATE INDEX order_lines_order_id_idx ON order_lines(order_id);
"""


def open_database(path: str | Path) -> sqlite3.Connection:
    """在交付连接前验证外键执行状态，避免调用方带着错误配置写入。"""
    connection = sqlite3.connect(path, isolation_level=None, timeout=5.0)
    try:
        connection.execute("PRAGMA foreign_keys = ON")
        state = connection.execute("PRAGMA foreign_keys").fetchone()
        if state != (1,):
            raise RuntimeError("SQLite 连接未启用外键约束")
    except (sqlite3.Error, RuntimeError):
        connection.close()
        raise
    return connection

这里显式使用 isolation_level=None，让此示例通过 SQL 的 BEGIN、COMMIT、ROLLBACK 管理事务；不要照搬到已有框架后再与框架事务管理混用。Python 3.12 以上若改用 autocommit=False，需要重新安排初始化时机，避免连接处于持续开启的事务中。事务模式见 Python 官方说明。

NOT NULL 表达“每条明细必须有订单”；仅有外键不能代替这个业务约束。子表索引用于降低按父键定位明细的成本，不决定外键是否开启。

调用方负责关闭连接；连接上下文管理器处理事务，不能替代关闭动作。这里用 closing 明确回收。生产中应让接口、批处理、迁移和运维工具使用同一连接工厂；使用连接池时，将初始化放在每个物理连接建立的钩子中。

4. 用回归测试锁住故障边界

将以下代码追加到文件末尾，执行 python -m unittest -v test_sqlite_foreign_keys.py。

class ForeignKeyTests(unittest.TestCase):
    """覆盖正常关联、失败写入、连接隔离和历史违规检查。"""

    def setUp(self) -> None:
        self.connection = open_database(":memory:")
        self.addCleanup(self.connection.close)
        self.connection.executescript(SCHEMA)

    def test_valid_reference(self) -> None:
        self.connection.execute("INSERT INTO orders VALUES (?)", (1,))
        self.connection.execute("INSERT INTO order_lines VALUES (?, ?)", (1, 1))
        self.assertEqual(
            self.connection.execute("SELECT order_id FROM order_lines").fetchall(),
            [(1,)],
        )

    def test_orphan_is_rejected(self) -> None:
        with self.assertRaises(sqlite3.IntegrityError):
            self.connection.execute(
                "INSERT INTO order_lines VALUES (?, ?)", (1, 999)
            )

    def test_null_is_rejected(self) -> None:
        with self.assertRaises(sqlite3.IntegrityError):
            self.connection.execute(
                "INSERT INTO order_lines VALUES (?, ?)", (1, None)
            )

    def test_referenced_parent_cannot_be_deleted(self) -> None:
        self.connection.execute("INSERT INTO orders VALUES (?)", (1,))
        self.connection.execute("INSERT INTO order_lines VALUES (?, ?)", (1, 1))
        with self.assertRaises(sqlite3.IntegrityError):
            self.connection.execute("DELETE FROM orders WHERE id = ?", (1,))

    def test_setting_inside_transaction_has_no_effect(self) -> None:
        with closing(sqlite3.connect(":memory:", isolation_level=None)) as raw:
            raw.execute("PRAGMA foreign_keys = OFF")
            raw.execute("BEGIN")
            raw.execute("PRAGMA foreign_keys = ON")
            self.assertEqual(raw.execute("PRAGMA foreign_keys").fetchone(), (0,))
            raw.execute("ROLLBACK")
            raw.execute("PRAGMA foreign_keys = ON")
            self.assertEqual(raw.execute("PRAGMA foreign_keys").fetchone(), (1,))

    def test_setting_is_per_connection(self) -> None:
        with TemporaryDirectory() as directory:
            path = Path(directory) / "orders.db"
            with closing(open_database(path)) as first:
                first.executescript(SCHEMA)
                with closing(open_database(path)) as second:
                    second.execute("PRAGMA foreign_keys = OFF")
                    self.assertEqual(
                        first.execute("PRAGMA foreign_keys").fetchone(), (1,)
                    )
                    second.execute(
                        "INSERT INTO order_lines VALUES (?, ?)", (1, 999)
                    )
                    self.assertEqual(
                        first.execute("PRAGMA foreign_key_check").fetchall(),
                        [("order_lines", 1, "orders", 0)],
                    )

    def test_enabling_does_not_repair_existing_data(self) -> None:
        self.connection.execute("PRAGMA foreign_keys = OFF")
        self.connection.execute("INSERT INTO order_lines VALUES (?, ?)", (1, 999))
        self.connection.execute("PRAGMA foreign_keys = ON")
        self.assertEqual(
            self.connection.execute("PRAGMA foreign_key_check").fetchall(),
            [("order_lines", 1, "orders", 0)],
        )
        self.connection.execute("INSERT INTO orders VALUES (?)", (999,))
        self.assertEqual(
            self.connection.execute("PRAGMA foreign_key_check").fetchall(), []
        )

    def test_failed_batch_can_be_rolled_back(self) -> None:
        self.connection.execute("BEGIN")
        try:
            self.connection.execute("INSERT INTO orders VALUES (?)", (1,))
            with self.assertRaises(sqlite3.IntegrityError):
                self.connection.execute(
                    "INSERT INTO order_lines VALUES (?, ?)", (1, 999)
                )
        finally:
            self.connection.execute("ROLLBACK")
        self.assertEqual(
            self.connection.execute("SELECT COUNT(*) FROM orders").fetchone(),
            (0,),
        )

第六项使用同一个真实临时文件的两个连接，验证其中一个关闭外键后，另一个仍然开启；单纯在两个独立内存库中检查开关无法覆盖这个使用场景。最后一项验证调用方执行整批回滚；默认即时外键失败不应被理解成整个事务已经自动撤销。

5. 处理已存在的孤儿明细

上线连接工厂只阻止新的违规写入。对于历史数据，先备份数据库并暂停相关写入口，再执行检查，按业务主键取证，例如：

SELECT line.id, line.order_id
FROM order_lines AS line
LEFT JOIN orders AS parent ON parent.id = line.order_id
WHERE parent.id IS NULL;

这条查询只针对本文 order_id NOT NULL、整数主键的两表结构。不要把它直接推广为所有复合外键的检查器。

对每条违规记录核对导入批次和原始订单：确有订单遗漏时补齐真实订单；明细订单号写错时修正关联；确认无效且允许删除时才隔离或删除。不能为了让检查变绿而创建没有业务依据的父记录。修复应在事务中完成，提交前再次运行 foreign_key_check，仍有违规就回滚并保留待处理清单。

若开启后出现 foreign key mismatch，检查父键是否确为主键或匹配的唯一键，以及列组合、排序规则是否匹配。这属于模型声明问题，不能通过关闭检查解决。大库完整检查的耗时需要在副本评估，再安排维护窗口。

6. 验收与预防

本地已执行本文八项标准库测试，以及上面的 SQL 故障复现。它们验证 SQLite 实际行为，不依赖外部数据库服务。未接入真实 ORM、连接池、多进程写入或生产数据文件，这些集成场景需要上线前补验。

验收时至少检查三条路径：接口创建的新连接、导入脚本创建的新连接、服务重启后重建的连接。每条路径都应读取到 foreign_keys=1，并拒绝一条不存在父键的写入；这些负向测试在隔离测试库运行。存量库另需得到可解释的违规清单或空检查结果。

不要把 integrity_check 返回正常当作外键验收；外键违规应由 foreign_key_check 检查。部署时保留数据库备份和写入失败记录；新约束暴露旧调用缺陷时，应修复或暂停该入口，避免回退成全局关闭外键。

这类故障的排查顺序是：确认实际文件，读取故障连接的开关，核对声明，再检查历史数据。将初始化前移到每次物理连接创建，并保留失败写入测试，才能防止“迁移时配置正确，运行时约束失效”再次出现。