Chapter 20 · Production Capstone: Build and Operate a Complete SQLite Application Database
Implement Application Access, Transactions, JSON/Search Features, and Concurrency Handling
Build a Python data-access layer with bound statements, explicit transaction ownership, idempotent writes, RETURNING, controlled JSON, WAL initialization, bounded busy handling, and multi-connection tests.
One application contract, one transaction owner
The capstone uses Python because it is widely available and its standard sqlite3 module exposes the core ideas without a third-party dependency. The design remains driver-neutral: open/configure a connection, verify capabilities, bind values, own transactions explicitly, map rows, classify errors, keep transactions short, and close resources. An async framework around this code does not change SQLite's one-writer-per-file behavior.
Initialize every connection with required runtime/version checks, foreign keys, trusted-schema policy, busy handling, WAL, and deliberate synchronous policy.
Use bound parameters for every data value and keep identifiers/SQL structure program-owned.
Implement an idempotent inspection transaction with UPSERT-style conflict handling and RETURNING.
Use controlled JSON for variable measurements without turning the whole model into a document store.
Retry SQLITE_BUSY only around a bounded, idempotent business operation.
Test foreground/background concurrency with multiple file-backed connections and verify failure recovery.
Connection initialization is part of correctness
Connection-local settings do not magically carry to every future handle. The repository function establishes the assumptions that later methods depend on. This capstone keeps Python in SQLite autocommit mode and issues BEGIN IMMEDIATE/COMMIT itself so transaction ownership is explicit.
from __future__ import annotationsimport sqlite3from pathlib import PathMIN_SQLITE = (3, 38, 0) # STRICT + RETURNING + built-in JSON baseline.def open_db(path: str | Path) -> sqlite3.Connection: # Keep SQLite autocommit on and own BEGIN/COMMIT explicitly. con = sqlite3.connect(path, autocommit=True, timeout=2.0) con.row_factory = sqlite3.Row runtime = tuple(map(int, sqlite3.sqlite_version.split('.'))) if runtime < MIN_SQLITE: con.close() raise RuntimeError(f"SQLite {MIN_SQLITE}+ required; got {sqlite3.sqlite_version}") # Version alone is not enough: builds can omit optional/default features. try: json_ok = con.execute("SELECT json_valid('{\"probe\":1}')").fetchone()[0] except sqlite3.Error as exc: con.close() raise RuntimeError("required SQLite JSON capability unavailable") from exc if json_ok != 1: con.close() raise RuntimeError("required SQLite JSON capability failed probe") # Connection policy: fail if required guarantees cannot be established. con.execute("PRAGMA foreign_keys = ON") if con.execute("PRAGMA foreign_keys").fetchone()[0] != 1: con.close() raise RuntimeError("foreign key enforcement unavailable") con.execute("PRAGMA trusted_schema = OFF") con.execute("PRAGMA busy_timeout = 2000") mode = con.execute("PRAGMA journal_mode = WAL").fetchone()[0] if str(mode).lower() != "wal": con.close() raise RuntimeError(f"WAL policy could not be enabled: {mode}") # FieldNotes values power-loss durability over the lower commit latency # of WAL+NORMAL, so this capstone chooses FULL deliberately. con.execute("PRAGMA synchronous = FULL") return conChecking sqlite3.sqlite_version matters because Python itself and the SQLite library are separately versioned. The local course runtime may differ from the current 3.53.4 baseline. Production startup should log both application version and sqlite_version()/sqlite_source_id() without logging sensitive SQL values.
Why the core feature baseline is 3.38.0
| Feature used by capstone | First broadly relevant SQLite version | Policy |
|---|---|---|
| Generated columns | 3.31.0 | Required by protocol projection. |
| RETURNING | 3.35.0 | Required by record_inspection result path. |
| STRICT tables | 3.37.0 | Required by schema. |
| JSON built in by default | 3.38.0 | Capstone declares 3.38.0+ and separately probes JSON capability. |
| WAL | Much older | Required deployment policy but filesystem suitability must still be validated. |
| FTS5 | Build option/module | Not required by this capstone; do not make core availability depend on it. |
A current patched release such as 3.53.4 remains the qualified target. “Minimum feature version” is not permission to deploy an obsolete vulnerable build indefinitely.
The business operation: inspection + note + outbox
The critical transaction has a clear invariant: if the inspection commits, the local sync event commits in the same transaction. The HTTP synchronization itself occurs later, outside the transaction. request_id is generated before the first attempt and stays stable across retries.
import jsonimport sqlite3import timeimport uuidclass FieldNotesBusy(RuntimeError): passclass FieldNotesValidation(ValueError): passdef record_inspection_once( con, *, request_id, device_id, technician_name, started_at, finished_at, outcome, summary, measurements, severity="info",): if outcome not in {"pass", "follow_up", "failed"}: raise FieldNotesValidation("invalid outcome") if severity not in {"info", "warning", "critical"}: raise FieldNotesValidation("invalid severity") payload = json.dumps(measurements, separators=(",", ":"), sort_keys=True) event_id = f"inspection:{request_id}" try: con.execute("BEGIN IMMEDIATE") inserted = con.execute( """ INSERT INTO inspection( device_id, request_id, technician_name, started_at, finished_at, outcome, summary, measurements_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(request_id) DO NOTHING RETURNING inspection_id, request_id, outcome """, (device_id, request_id, technician_name, started_at, finished_at, outcome, summary, payload), ).fetchone() if inserted is None: # Idempotent replay: return the already committed operation. existing = con.execute( "SELECT inspection_id, request_id, outcome " "FROM inspection WHERE request_id=?", (request_id,), ).fetchone() con.execute("COMMIT") return dict(existing), False inspection_id = inserted["inspection_id"] if summary: con.execute( """ INSERT INTO maintenance_note( device_id, inspection_id, occurred_at, severity, note_text ) VALUES (?, ?, ?, ?, ?) """, (device_id, inspection_id, finished_at or started_at, severity, summary), ) con.execute( """ INSERT INTO sync_outbox(event_id, aggregate_type, aggregate_id, payload_json) VALUES (?, 'inspection', ?, ?) """, (event_id, inspection_id, json.dumps({"request_id": request_id, "inspection_id": inspection_id}, separators=(",", ":"))), ) con.execute("COMMIT") return dict(inserted), True except Exception: if con.in_transaction: con.execute("ROLLBACK") raisedef record_inspection(con, **kwargs): # Retry the whole idempotent business operation, not an arbitrary SQL line. delays = (0.05, 0.15, 0.35) for attempt, delay in enumerate(delays, start=1): try: return record_inspection_once(con, **kwargs) except sqlite3.OperationalError as exc: code = getattr(exc, "sqlite_errorcode", None) if code is None or (code & 0xFF) != sqlite3.SQLITE_BUSY: raise if attempt == len(delays): raise FieldNotesBusy("write remained busy after bounded retries") from exc time.sleep(delay)The statement uses ON CONFLICT(request_id) DO NOTHING RETURNING .... A first execution receives the new row. A replay receives no RETURNING row and selects the already committed inspection. The retry wrapper restarts the whole idempotent operation, not just whichever SQL line happened to encounter BUSY.
JSON is solving a real variability problem
Different equipment families emit different measurements. Turning every possible sensor key into a nullable column would produce schema churn; storing stable facts such as site, device, outcome, status, time, and identity inside JSON would hide important relational structure. The capstone therefore uses a hybrid.
import jsonmetadata = {"protocol": "modbus", "firmware": "3.8.2", "ports": [1, 2]}con.execute( """ INSERT INTO device(site_id, device_code, device_name, metadata_json) VALUES (?, ?, ?, ?) """, (1, "PUMP-007", "Cooling Pump 007", json.dumps(metadata, separators=(",", ":"))),)row = con.execute( "SELECT device_code, protocol FROM device WHERE device_code=?", ("PUMP-007",),).fetchone()print(dict(row))# {'device_code': 'PUMP-007', 'protocol': 'modbus'}The generated protocol column gives the common query a stable typed interface. If another JSON property becomes operationally important, promote it intentionally rather than spraying expression paths throughout application code.
Why FTS5 is not in the required core
Maintenance-note text could eventually benefit from full-text search, but the current requirements need recent notes by device, not relevance-ranked global text search. Adding FTS5 would create compile-option and synchronization obligations without satisfying a current acceptance criterion. This is production restraint: Chapter 14 taught a capability; the capstone uses it only when the product needs it.
An optional exercise in Lesson 5 adds FTS5 behind runtime capability detection while preserving the repository interface. It is intentionally not a core startup dependency.
Structured error handling
Application code should distinguish validation failures, expected constraint failures, transient busy conditions, read-only/full/I/O failures, and corruption signals. Do not parse entire English error strings as a stable machine interface when the driver exposes SQLite result codes.
def classify_sqlite_error(exc: sqlite3.Error) -> str: code = getattr(exc, "sqlite_errorcode", None) if code is not None and (code & 0xFF) == sqlite3.SQLITE_BUSY: return "transient_busy" if code == sqlite3.SQLITE_CONSTRAINT: return "constraint" if code == sqlite3.SQLITE_READONLY: return "read_only" if code == sqlite3.SQLITE_FULL: return "storage_full" if code == sqlite3.SQLITE_CORRUPT: return "corrupt" return "sqlite_error"Extended result codes may add detail and drivers differ in how they expose them. Log the primary/extended code, operation name, request identifier, and runtime version; do not log sensitive note text or secret-bearing payloads by default.
Controlled two-connection concurrency test
Use a file-backed database because :memory: would create separate private databases for ordinary independent connections and would not exercise file locks/WAL. Connection A holds a short write transaction. Connection B has a deliberately tiny busy timeout for the test and proves the conflict is observable.
import sqlite3from pathlib import Pathpath = Path("fieldnotes-capstone.sqlite")a = open_db(path)b = open_db(path)b.execute("PRAGMA busy_timeout=50")a.execute("BEGIN IMMEDIATE")a.execute("UPDATE device SET status='inspection_due' WHERE device_id=1")try: b.execute("BEGIN IMMEDIATE")except sqlite3.OperationalError as exc: assert getattr(exc, "sqlite_errorcode", None) == sqlite3.SQLITE_BUSYfinally: a.execute("ROLLBACK")# After the first writer releases the lock, B can acquire it.b.execute("BEGIN IMMEDIATE")b.execute("ROLLBACK")a.close(); b.close()Failure injection: atomicity, not optimism
Place an intentional exception after the inspection INSERT but before the outbox INSERT. The repository's exception path rolls the transaction back. Verification must show zero new inspection and zero new outbox row for that request.
con.execute("BEGIN IMMEDIATE")try: con.execute( "INSERT INTO inspection(device_id, request_id, technician_name, " "started_at, outcome, summary) VALUES (?, ?, ?, ?, ?, ?)", (1, "fail-demo", "Tech A", "2026-08-12T08:00:00Z", "pass", "demo"), ) raise RuntimeError("injected failure before outbox") # INSERT sync_outbox would have happened here. con.execute("COMMIT")except Exception: if con.in_transaction: con.execute("ROLLBACK")assert con.execute( "SELECT count(*) FROM inspection WHERE request_id='fail-demo'").fetchone()[0] == 0assert con.execute( "SELECT count(*) FROM sync_outbox WHERE event_id='inspection:fail-demo'").fetchone()[0] == 0Equivalent binding beneath different languages
The syntax changes; the contract does not. A Node/.NET/Java implementation should preserve the same request-id idempotency, short transaction boundary, connection initialization, and result-code handling rather than “translating” only the SQL strings.
connection = open_and_verify_database()statement = prepare("SELECT ... WHERE device_code = ?")statement.bind(device_code)row = statement.step_and_map()begin_immediate()try: insert_inspection_with_bound_values() insert_outbox_with_bound_values() commit()except: rollback_if_active() classify_and_propagate_error()Application checkpoint
Own the operation, not just the cursor
Reason about retries and boundaries.
- Why does record_inspection retry the whole operation instead of only the failed INSERT?
- Why must request_id be generated before retry attempts?
- Why is the HTTP sync call after COMMIT rather than before it?
- What would be wrong with storing site_id and device status only inside measurements_json?
- Why does enabling WAL not remove the need for SQLITE_BUSY handling?
- Why is a file-backed test required for locking behavior?
Review the answers
The business operation is the atomic unit and may have executed partially before an error; restarting it safely requires idempotency. A stable request_id lets the database recognize replays. Remote calls inside transactions make lock duration depend on the network and cannot be rolled back by SQLite. Stable relational facts belong in columns/relationships so constraints and indexes can enforce them. WAL still allows only one writer per database file. File-backed independent connections are needed to exercise the pager/VFS locking and WAL behavior that an isolated in-memory database does not reproduce.
Bridge to operations acceptance
The application path now works in the happy case and in injected failure/lock conflicts. Lesson 4 refuses to call that “production ready” until tests, representative benchmarks, backup/restore, integrity checks, and security acceptance all pass.