Python SQLite 报 database is locked：有限等待、短事务与故障测试

python3 --version
python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
mkdir sqlite-lock-demo
cd sqlite-lock-demo
python3 - <<'PY_SETUP'
import sqlite3
from contextlib import closing
from pathlib import Path

db = Path("demo.sqlite")
if db.exists():
    raise SystemExit("Use a fresh demo directory")
with closing(sqlite3.connect(db, isolation_level=None)) as con:
    con.execute("CREATE TABLE results (job TEXT PRIMARY KEY, value TEXT NOT NULL)")
print(db.resolve())
PY_SETUP

"""Save one prepared result in an existing SQLite database. Python 3.11+."""
import argparse
from contextlib import closing
import math
from pathlib import Path
import sqlite3
import sys


def wait_seconds(text):
    value = float(text)
    if not math.isfinite(value) or not 0 <= value <= 30:
        raise argparse.ArgumentTypeError("timeout must be finite and in 0..30 seconds")
    return value


def save_result(database, job, value, *, timeout=1.0):
    path = Path(database).resolve(strict=True)
    # mode=rw prevents accidental creation if the path disappears before opening.
    with closing(sqlite3.connect(
        path.as_uri() + "?mode=rw", uri=True,
        timeout=timeout, isolation_level=None,
    )) as con:
        try:
            con.execute("BEGIN IMMEDIATE")
            con.execute(
                "INSERT INTO results(job, value) VALUES (?, ?) "
                "ON CONFLICT(job) DO UPDATE SET value=excluded.value",
                (job, value),
            )
            con.execute("COMMIT")
        except BaseException:
            if con.in_transaction:
                con.execute("ROLLBACK")
            raise


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("database", type=Path)
    parser.add_argument("job")
    parser.add_argument("value")
    parser.add_argument("--timeout", type=wait_seconds, default=1.0)
    args = parser.parse_args()
    try:
        save_result(args.database, args.job, args.value, timeout=args.timeout)
    except (OSError, sqlite3.Error) as exc:
        name = getattr(exc, "sqlite_errorname", type(exc).__name__)
        print(f"ERROR: {name}: {exc}; db={args.database.resolve()}", file=sys.stderr)
        return 1
    print(f"OK: job={args.job}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 save_result.py --help
python3 save_result.py demo.sqlite daily ready --timeout 1
python3 save_result.py demo.sqlite daily updated --timeout 1
python3 - <<'PY_CHECK'
import sqlite3
from contextlib import closing

with closing(sqlite3.connect("file:demo.sqlite?mode=ro", uri=True)) as con:
    print(con.execute("SELECT job, value FROM results ORDER BY job").fetchall())
PY_CHECK

OK: job=daily
OK: job=daily
[('daily', 'updated')]

"""Hold a real write lock while a separate process attempts to write."""
from contextlib import closing
from pathlib import Path
import sqlite3
import subprocess
import sys

from save_result import save_result

db = Path("demo.sqlite").resolve(strict=True)
command = [sys.executable, "save_result.py", str(db), "daily", "after-lock",
           "--timeout", "0.2"]
with closing(sqlite3.connect(db, isolation_level=None)) as holder:
    holder.execute("BEGIN IMMEDIATE")
    try:
        holder.execute("INSERT INTO results VALUES ('uncommitted', 'discard-me')")
        blocked = subprocess.run(command, capture_output=True, text=True, timeout=10)
        assert blocked.returncode == 1, blocked
        assert "SQLITE_BUSY" in blocked.stderr, blocked.stderr
        assert blocked.stdout == "", blocked.stdout
        print("PASS: competing writer returned 1 with SQLITE_BUSY")
    finally:
        holder.execute("ROLLBACK")

released = subprocess.run(command, capture_output=True, text=True, timeout=10)
assert released.returncode == 0, released.stderr
assert released.stdout.strip() == "OK: job=daily", released.stdout

with closing(sqlite3.connect(db, isolation_level=None)) as con:
    assert con.execute("SELECT value FROM results WHERE job='daily'").fetchone() == (
        "after-lock",
    )
    assert con.execute("SELECT count(*) FROM results WHERE job='uncommitted'").fetchone() == (0,)

try:
    save_result(db, "daily", None)
except sqlite3.IntegrityError:
    pass
else:
    raise AssertionError("NOT NULL constraint should fail")

save_result(db, "daily", "recovered")
with closing(sqlite3.connect(db)) as con:
    assert con.execute("SELECT job, value FROM results").fetchall() == [("daily", "recovered")]
print("PASS: release, rollback, repeat update, and final rows verified")

python3 test_lock.py

PASS: competing writer returned 1 with SQLITE_BUSY
PASS: release, rollback, repeat update, and final rows verified