接口返回成功，主表记录也还在，但关联明细突然清空。遇到这种现象，先找到实际执行的 SQL：如果所谓“更新”用了 INSERT OR REPLACE，就要沿着删除路径排查。
本文用一个订单与明细模型复现问题，再把写入改成有明确冲突目标的 UPSERT。实验只使用内存数据库，不连接生产服务。
适用场景与现象
适用于 SQLite 本地业务库、桌面应用或同步程序：主表存在唯一业务键，明细通过外键关联，删除主表时按业务设计级联删除明细。同步程序希望“存在则更新，不存在则插入”。
典型现象是：同步前订单有明细，同步后订单名称更新成功、明细数量变为零，没有 SQL 异常。甚至重新写回相同的主键也不能避免这个问题。
这篇文章讨论的是冲突写入的删除语义。连接已开启外键校验，外键本身按设计工作。
原因：REPLACE 会删除发生唯一冲突的旧行
SQLite 的 REPLACE 在主键或唯一键冲突时，会先删除导致冲突的已有行，再继续写入。它不能等同于原地修改字段。这个删除步骤可能触发明细表的 ON DELETE CASCADE。详见 SQLite 冲突处理文档。
因此需要分别确认两个问题：写入是否替换了主表行；外键的删除动作是否清理了关联行。最终主键值相同，也不能证明中间没有发生删除。
用一个可重复实验把链路跑出来
将下面完整代码保存为 sqlite_replace_demo.py，使用 Python 3.10 以上运行；底层 SQLite 需至少 3.24.0。代码每次建立独立内存库，不修改磁盘文件。
import sqlite3
import unittest
from contextlib import closing
SCHEMA = """
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
order_key TEXT NOT NULL UNIQUE,
display_name TEXT NOT NULL
);
CREATE TABLE order_items (
id INTEGER PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
quantity INTEGER NOT NULL CHECK (quantity > 0)
);
CREATE INDEX order_items_order_id_idx ON order_items(order_id);
"""
UPSERT = """
INSERT INTO orders (order_key, display_name)
VALUES (?, ?)
ON CONFLICT (order_key) DO UPDATE
SET display_name = excluded.display_name
"""
def create_database() -> sqlite3.Connection:
"""创建独立实验库，先开启外键，再创建并填充数据。"""
connection = sqlite3.connect(":memory:")
connection.execute("PRAGMA foreign_keys = ON")
if connection.execute("PRAGMA foreign_keys").fetchone() != (1,):
connection.close()
raise RuntimeError("实验连接未开启外键校验")
connection.executescript(SCHEMA)
with connection:
connection.execute(
"INSERT INTO orders VALUES (?, ?, ?)", (10, "order-a", "原名称")
)
connection.execute("INSERT INTO order_items VALUES (?, ?, ?)", (1, 10, 2))
return connection
def read_state(connection: sqlite3.Connection) -> tuple[list[tuple], list[tuple]]:
"""同时读取主表身份和明细，避免只检查写入返回值。"""
orders = connection.execute(
"SELECT id, order_key, display_name FROM orders ORDER BY id"
).fetchall()
items = connection.execute(
"SELECT id, order_id, quantity FROM order_items ORDER BY id"
).fetchall()
return orders, items
class ReplaceTests(unittest.TestCase):
def test_replace_same_primary_key_deletes_items(self) -> None:
with closing(create_database()) as connection:
with connection:
connection.execute(
"INSERT OR REPLACE INTO orders VALUES (?, ?, ?)",
(10, "order-a", "新名称"),
)
self.assertEqual(read_state(connection), ([(10, "order-a", "新名称")], []))
self.assertEqual(
connection.execute("PRAGMA foreign_key_check").fetchall(), []
)
def test_upsert_preserves_identity_and_items(self) -> None:
with closing(create_database()) as connection:
with connection:
connection.execute(UPSERT, ("order-a", "新名称"))
self.assertEqual(
read_state(connection),
([(10, "order-a", "新名称")], [(1, 10, 2)]),
)
def test_upsert_inserts_new_order(self) -> None:
with closing(create_database()) as connection:
with connection:
connection.execute(UPSERT, ("order-b", "新订单"))
orders, items = read_state(connection)
self.assertEqual(len(orders), 2)
self.assertEqual(orders[0], (10, "order-a", "原名称"))
self.assertEqual(orders[1][1:], ("order-b", "新订单"))
self.assertEqual(items, [(1, 10, 2)])
def test_repeat_upsert_keeps_one_order(self) -> None:
with closing(create_database()) as connection:
with connection:
for _ in range(3):
connection.execute(UPSERT, ("order-a", "相同名称"))
self.assertEqual(
read_state(connection),
([(10, "order-a", "相同名称")], [(1, 10, 2)]),
)
def test_invalid_input_does_not_damage_existing_data(self) -> None:
with closing(create_database()) as connection:
before = read_state(connection)
with self.assertRaises(sqlite3.IntegrityError):
with connection:
connection.execute(UPSERT, ("order-a", None))
self.assertEqual(read_state(connection), before)
def test_rollback_restores_replaced_order_and_items(self) -> None:
with closing(create_database()) as connection:
before = read_state(connection)
connection.execute("BEGIN")
connection.execute(
"INSERT OR REPLACE INTO orders VALUES (?, ?, ?)",
(10, "order-a", "待回滚名称"),
)
self.assertEqual(read_state(connection)[1], [])
connection.rollback()
self.assertEqual(read_state(connection), before)
if __name__ == "__main__":
print(f"SQLite 运行库版本：{sqlite3.sqlite_version}")
unittest.main(verbosity=2)
执行命令：
python .\sqlite_replace_demo.py
第一项测试确认故障确实发生：显式写回主键 10，订单仍为 10，明细却为空。第二项测试验证修复后订单身份及明细同时保留。剩余测试检查新增、重复同步、无效输入，以及未提交时的回滚恢复。
注意第一项测试中的 foreign_key_check 返回空列表：级联删除后没有孤儿行，数据库仍满足外键约束。这项检查不能证明业务明细没有丢失，必须另外核对业务数量或具体记录。
现场排查：先核对 SQL，再核对约束
在实际发生写入的连接上执行以下只读诊断语句；表名按真实结构调整：
SELECT sqlite_version();
PRAGMA foreign_keys;
PRAGMA foreign_key_list(order_items);
PRAGMA index_list(orders);
SELECT type, name, sql
FROM sqlite_schema
WHERE tbl_name IN ('orders', 'order_items')
ORDER BY type, name;
SELECT id, order_key, display_name FROM orders WHERE order_key = 'order-a';
SELECT id, order_id, quantity FROM order_items WHERE order_id = 10;
PRAGMA foreign_key_check;
重点看 foreign_keys 是否为 1，以及 foreign_key_list 的 on_delete 是否为 CASCADE。同时检查表和触发器定义，确认是否还有其他删除路径。SQLite 外键动作的语义见 外键文档。
从应用仓储方法、ORM 生成 SQL 或已有的脱敏 SQL 记录中定位 REPLACE、INSERT OR REPLACE；不要只依据方法名中的 save、update 判断执行语义。现场日志不要打印原始订单内容或敏感参数。
修复：明确冲突键，只更新允许覆盖的字段
示例 UPSERT 以 order_key 为业务身份，冲突时只更新 display_name。excluded.display_name 表示本次待写入的新值，原有 id 保留，因此这个更新不会走父行删除路径。语法及约束范围见 SQLite UPSERT 文档。
迁移前核对以下条件：
order_key确实有唯一约束。如果业务身份由租户和订单号共同组成，应建立对应联合唯一约束，并让冲突目标完全匹配。- 不把主键、创建时间等字段顺手放入更新列表。实际业务中的状态字段还需要独立的状态流转校验。
- 保留参数绑定；不要把外部输入拼接进 SQL。接口层还应按业务限制字段类型、长度和权限。
- 其他唯一约束、非空约束及外键错误仍需处理。UPSERT 不能当作通用的“忽略所有错误”。
如果业务明确要求“必须已有记录才能修改”，参数化 UPDATE ... WHERE ... 更符合意图，再依据受影响行数处理不存在的订单。不要先查有没有、再在没有事务保障的情况下决定插入或更新。
已丢失数据怎样处理
事务尚未提交时，可以像第六项测试一样回滚，主表变化与级联删除一起撤销。提交后再调用 rollback() 不能恢复之前的明细。
已经提交的事故应先停止触发问题的同步入口，保留现场并制作可恢复备份，再从经过校验的备份或权威上游恢复明细。恢复时核对订单身份、明细数量和业务汇总，并执行外键检查；先在副本演练，避免重复导入。不要凭主表金额推造明细。
发布修复时先在数据库副本运行回归，再用少量真实结构的数据观察。代码回退如果重新启用 REPLACE，会重新引入删除风险；回退期间应暂停相关写入入口，而不是继续使用旧逻辑同步。
防止复发与验证边界
代码评审中检查主表写入是否出现替换语句；回归断言同时检查主键、关联行和业务字段，而不是只判断“调用未抛异常”。保留正常删除订单时的级联设计，不要通过关闭外键掩盖错误写法。
本文在 Python 3.14.6、SQLite 3.50.4 上执行上述六项测试，使用独立内存数据库。它验证了 SQL 语义及事务内回滚，没有验证生产 ORM、磁盘恢复、并发同步或其他数据库的同名语法。接入真实项目后，还应覆盖该项目的触发器、其他唯一键及实际事务边界。
问题的关键是写入语句表达的业务动作：保持订单身份的字段更新，应采用明确的更新语义，并把明细保留作为回归条件。
Discussion
评论