SQLite 事务内批量 Upsert

功能说明

这段代码演示如何把一批库存记录安全地写入 SQLite：已存在的 SKU 更新数量与价格，新 SKU 插入，所有行要么全部成功，要么全部回滚。它适合离线同步、批量导入或定时任务，避免“写了一半失败”造成数据不一致。

实现把校验放在事务之前，任何非法行都会在开启事务前被拒绝；真正的写入使用 executemany 与 ON CONFLICT ... DO UPDATE，一条语句完成插入或更新，不需要先查询再分支。

关键代码

SCHEMA 用 CHECK 约束保证数量和价格非负，这是数据库层的最后一道防线。UPSERT 以 sku 为冲突键，用 excluded.quantity 和 excluded.price_cents 覆盖旧值。

apply_stock_updates 先逐行检查空 SKU、负数量、负价格，然后执行 BEGIN IMMEDIATE 获取写锁；executemany 成功后 commit，任何异常都会 rollback 并重新抛出。返回值是 cursor.rowcount，可用于日志或指标。

使用注意

SQLite 的连接默认隔离级别与自动提交行为容易让人误解，这里显式 BEGIN IMMEDIATE，避免多个写入者同时升级锁导致 database is locked。不要在事务中做网络请求或长时间计算，否则会阻塞其他写入。

所有值都使用 ? 占位符，绝不能把 SKU 拼进 SQL。rowcount 在不同 SQLite 版本和语句类型下语义略有差异，不要把它当作精确的业务计数。批量很大时应分批提交，控制单次事务的锁持有时间和内存占用。

验证方式

第一次写入两行返回 2；第二次更新 A-1 并插入 C-3 也返回 2。最终查询得到 A-1 数量 9、价格 1150，B-2 保持 3/800，C-3 为 1/450，共三行。传入负数量会抛出 ValueError，断言行数仍为 3，证明坏批次没有污染已有数据。

代码示例

from __future__ import annotations

import sqlite3

SCHEMA = """
CREATE TABLE inventory (
    sku TEXT PRIMARY KEY,
    quantity INTEGER NOT NULL CHECK (quantity >= 0),
    price_cents INTEGER NOT NULL CHECK (price_cents >= 0)
)
"""

UPSERT = """
INSERT INTO inventory (sku, quantity, price_cents)
VALUES (?, ?, ?)
ON CONFLICT(sku) DO UPDATE SET
    quantity = excluded.quantity,
    price_cents = excluded.price_cents
"""


def apply_stock_updates(
    connection: sqlite3.Connection,
    rows: list[tuple[str, int, int]],
) -> int:
    for sku, quantity, price_cents in rows:
        if not sku or quantity < 0 or price_cents < 0:
            raise ValueError(f"invalid inventory row: {sku!r}")

    connection.execute("BEGIN IMMEDIATE")
    try:
        cursor = connection.executemany(UPSERT, rows)
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    return cursor.rowcount


def main() -> None:
    connection = sqlite3.connect(":memory:")
    connection.executescript(SCHEMA)

    first = apply_stock_updates(connection, [("A-1", 5, 1200), ("B-2", 3, 800)])
    second = apply_stock_updates(connection, [("A-1", 9, 1150), ("C-3", 1, 450)])
    print("affected rows:", first, second)
    print(connection.execute(
        "SELECT sku, quantity, price_cents FROM inventory ORDER BY sku"
    ).fetchall())

    try:
        apply_stock_updates(connection, [("D-4", -1, 100)])
    except ValueError as exc:
        print(f"rejected: {exc}")
    assert connection.execute("SELECT COUNT(*) FROM inventory").fetchone()[0] == 3


if __name__ == "__main__":
    main()