Python SQLite UPSERT vs REPLACE: Keep Related Records When Updating

python3 --version
python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
mkdir sqlite-upsert-demo
cd sqlite-upsert-demo

"""Compare REPLACE and UPSERT in a fresh in-memory database. Python 3.11+."""
import argparse
from contextlib import closing
import json
import sqlite3
import sys


STATEMENTS = {
    "replace": """INSERT OR REPLACE INTO jobs(name, status, created_at)
                  VALUES (?, ?, ?)""",
    "upsert": """INSERT INTO jobs(name, status, created_at) VALUES (?, ?, ?)
                 ON CONFLICT(name) DO UPDATE SET status = excluded.status""",
}


def open_demo():
    if sqlite3.sqlite_version_info < (3, 24, 0):
        raise RuntimeError("SQLite 3.24.0+ is required")
    options = {"isolation_level": None}
    if sys.version_info >= (3, 12):
        options["autocommit"] = True
    con = sqlite3.connect(":memory:", **options)
    try:
        con.execute("PRAGMA foreign_keys = ON")
        if con.execute("PRAGMA foreign_keys").fetchone() != (1,):
            raise RuntimeError("foreign keys are not enabled")
        con.execute("""CREATE TABLE jobs (
            name TEXT PRIMARY KEY NOT NULL,
            status TEXT NOT NULL,
            created_at TEXT NOT NULL
        )""")
        con.execute("""CREATE TABLE results (
            id INTEGER PRIMARY KEY,
            job_name TEXT NOT NULL REFERENCES jobs(name) ON DELETE CASCADE,
            value TEXT NOT NULL
        )""")
        con.execute("INSERT INTO jobs VALUES (?, ?, ?)",
                    ("nightly", "queued", "2026-01-01"))
        con.execute("INSERT INTO results VALUES (?, ?, ?)",
                    (1, "nightly", "report.csv"))
        return con
    except Exception:
        con.close()
        raise


def write_job(con, mode, name, status, created_at):
    statement = STATEMENTS[mode]
    con.execute("BEGIN")
    try:
        con.execute(statement, (name, status, created_at))
        con.execute("COMMIT")
    except Exception:
        if con.in_transaction:
            con.execute("ROLLBACK")
        raise


def snapshot(con):
    return {
        "jobs": [list(row) for row in con.execute(
            "SELECT name, status, created_at FROM jobs ORDER BY name")],
        "results": [list(row) for row in con.execute(
            "SELECT id, job_name, value FROM results ORDER BY id")],
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--mode", choices=STATEMENTS, default="upsert")
    args = parser.parse_args()
    try:
        with closing(open_demo()) as con:
            before = snapshot(con)
            write_job(con, args.mode, "nightly", "done", "2026-02-01")
            result = {"mode": args.mode, "before": before, "after": snapshot(con)}
        print(json.dumps(result, ensure_ascii=True, sort_keys=True))
        return 0
    except (sqlite3.Error, RuntimeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

python3 upsert_demo.py --help
python3 upsert_demo.py --mode replace
python3 upsert_demo.py --mode upsert

"""Run beside upsert_demo.py; all databases are disposable and in-memory."""
from contextlib import closing
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
from upsert_demo import open_demo, snapshot, write_job

script = Path(__file__).with_name("upsert_demo.py")
for mode in ("replace", "upsert"):
    run = subprocess.run([sys.executable, str(script), "--mode", mode],
                         text=True, capture_output=True, check=True)
    assert run.stderr == ""
    report = json.loads(run.stdout)
    assert report["before"]["results"] == [[1, "nightly", "report.csv"]]
    after = report["after"]
    assert after["jobs"][0][:2] == ["nightly", "done"]
    if mode == "replace":
        assert after["results"] == []
        assert after["jobs"][0][2] == "2026-02-01"
    else:
        assert after["results"] == report["before"]["results"]
        assert after["jobs"][0][2] == "2026-01-01"

with closing(open_demo()) as con:
    # New key inserts, the same key updates only the requested field.
    name = "daily'中文"
    write_job(con, "upsert", name, "queued", "2026-03-01")
    write_job(con, "upsert", name, "done", "2099-01-01")
    write_job(con, "upsert", "nightly", "done", "2099-01-01")
    write_job(con, "upsert", "nightly", "done", "2099-01-01")
    before = snapshot(con)
    assert before["jobs"] == [
        [name, "done", "2026-03-01"],
        ["nightly", "done", "2026-01-01"],
    ]
    assert before["results"] == [[1, "nightly", "report.csv"]]
    try:
        write_job(con, "upsert", "nightly", None, "2099-01-01")
    except sqlite3.IntegrityError:
        pass
    else:
        raise AssertionError("NOT NULL failure was expected")
    assert snapshot(con) == before
    assert not con.in_transaction
    assert con.execute("PRAGMA foreign_key_check").fetchall() == []

for args, expected in [(["--help"], 0), (["--mode", "unknown"], 2)]:
    run = subprocess.run([sys.executable, str(script), *args], capture_output=True)
    assert run.returncode == expected
print("PASS: REPLACE loss, UPSERT preservation, insert, repeat, rollback, CLI")

python3 test_upsert.py

PASS: REPLACE loss, UPSERT preservation, insert, repeat, rollback, CLI