Fix Python SQLite Foreign Keys: Connection Checks, Batch Rollback, and Orphan Audits

Your SQLite schema declares REFERENCES, yet a batch still inserts a result for a job that does not exist. The missing step may be enabling foreign-key checks on the current connection, not correcting the INSERT. This beginner-to-intermediate tutorial compares the same batch with checks off and on, verifies full rollback, and audits old orphan rows without touching a real database.

1. Method: declare the relationship, enable checks, then audit old data

A schema declares relationships; connection settings determine whether they are enforced. Do not depend on your SQLite build’s default: set PRAGMA foreign_keys = ON after connecting, immediately query it, and require 1. Connections created by a pool or another process need the same initialization. The SQLite enabling guide documents the per-connection setting and its transaction restriction.

The parent table jobs holds tasks; the child table results holds their outputs. results.job_id must reference an existing job. NOT NULL also makes that relationship mandatory. This example does not request cascading deletion: deleting a referenced parent should fail.

This is neither a backup replacement nor an automatic repair tool. Before changing real data, perform a SQLite backup and restore check. Every database on this page exists only in memory.

2. Setup: use only the Python standard library

Use Python 3.11 or newer with the sqlite3 module available; no pip package or SQLite CLI is needed. These examples were tested on Linux with Python 3.11.4. Run the steps in order and stay in the new directory:

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

Save the next section as fk_demo.py, or download the complete demo script. It accepts --enforce on or --enforce off, defaulting to on; off is only a failure-control setting.

3. Implementation: set and verify the flag before BEGIN

"""Compare foreign-key enforcement in a disposable in-memory database."""
import argparse
import json
import sqlite3
import sys


def open_db(enforce=True):
    # No implicit BEGIN: all write transactions below are explicit.
    con = sqlite3.connect(":memory:", isolation_level=None)
    try:
        con.execute("PRAGMA foreign_keys = " + ("ON" if enforce else "OFF"))
        expected = (1 if enforce else 0,)
        if con.execute("PRAGMA foreign_keys").fetchone() != expected:
            raise RuntimeError("foreign-key setting was not applied")
        con.execute("CREATE TABLE jobs (id INTEGER PRIMARY KEY)")
        con.execute("""CREATE TABLE results (
            id INTEGER PRIMARY KEY,
            job_id INTEGER NOT NULL REFERENCES jobs(id)
        )""")
        con.execute("CREATE INDEX results_job_id ON results(job_id)")
        con.execute("INSERT INTO jobs VALUES (1)")
        return con
    except Exception:
        con.close()
        raise


def run_demo(enforce=True):
    con = open_db(enforce)
    error = None
    try:
        try:
            con.execute("BEGIN")
            con.execute("INSERT INTO results VALUES (?, ?)", (1, 1))
            con.execute("INSERT INTO results VALUES (?, ?)", (2, 999))
            con.execute("COMMIT")
        except sqlite3.IntegrityError as exc:
            con.execute("ROLLBACK")
            error = str(exc)
        report = {
            "accepted": error is None,
            "foreign_keys": con.execute("PRAGMA foreign_keys").fetchone()[0],
            "result_rows": con.execute("SELECT COUNT(*) FROM results").fetchone()[0],
            "violations": con.execute("PRAGMA foreign_key_check").fetchall(),
        }
        return report, error
    finally:
        con.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--enforce", choices=("on", "off"), default="on")
    args = parser.parse_args()
    report, error = run_demo(args.enforce == "on")
    print(json.dumps(report, sort_keys=True))
    if error:
        print("ERROR: " + error, file=sys.stderr)
        return 1
    return 0


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

isolation_level=None disables Python’s implicit transaction opening; the batch explicitly issues BEGIN, COMMIT, or ROLLBACK. See Python sqlite3 transaction control. This example does not use the autocommit parameter added in Python 3.12; do not mix in a connection policy that always keeps a transaction open.

The script creates job 1, then attempts (1, 1) and (2, 999) in one batch. The second row references a missing job. With enforcement on, exception handling explicitly rolls back the entire batch, including its first valid result. Failure of one statement does not by itself undo all preceding writes.

The results_job_id index supports finding children by parent key; it is not the enforcement switch. SQL values use ? binding instead of interpolation; the PRAGMA concatenation contains only fixed ON and OFF strings from the code.

4. Validation: compare rows and exit codes for the same input

python3 fk_demo.py --help
python3 fk_demo.py --enforce off
if python3 fk_demo.py --enforce on; then
  printf 'UNEXPECTED: invalid batch accepted\n' >&2
  exit 1
else
  rc=$?
  printf 'protected exit=%s\n' "$rc"
  test "$rc" -eq 1
fi

With enforcement off, stdout is the following, the exit code is 0, and stderr is empty:

{"accepted": true, "foreign_keys": 0, "result_rows": 2, "violations": [["results", 2, "jobs", 0]]}

With enforcement on, stdout is the following; stderr is ERROR: FOREIGN KEY constraint failed, the script exits 1, and the shell then prints protected exit=1:

{"accepted": false, "foreign_keys": 1, "result_rows": 0, "violations": []}

result_rows: 0 demonstrates full-batch rollback. accepted: false is still a failure report: JSON on stdout does not imply success. Automation must check the exit code too. An invalid --enforce value returns argparse’s exit code 2; help returns 0.

Download the regression tests, place them beside the script, and run:

python3 test_fk_demo.py

They check exit codes, row counts, stderr, historical violations, an ineffective transaction-time toggle, and rejection of referenced-parent deletion. They never connect to production or create database files.

5. Audit historical violations: enabling is not repairing

Save this as legacy_check.py and run python3 legacy_check.py in the demo directory:

from fk_demo import open_db

con = open_db(False)  # Reproduce a legacy writer only; not a production setting.
try:
    con.execute("INSERT INTO results VALUES (7, 999)")
    con.execute("PRAGMA foreign_keys = ON")
    print("enabled:", con.execute("PRAGMA foreign_keys").fetchone()[0])
    print("violations:", con.execute("PRAGMA foreign_key_check").fetchall())
finally:
    con.close()

Output:

enabled: 1
violations: [('results', 7, 'jobs', 0)]

The PRAGMA reference defines the four fields: child table, offending rowid, parent table, and foreign-key index. Enabling checks neither removes nor repairs existing orphans; 7 identifies the offending row in this ordinary rowid table.

For a legacy migration, run PRAGMA foreign_key_check against a stable backup or a copy with writers stopped. Nonempty results should block acceptance. Use business evidence to restore parents, correct references, or quarantine bad data rather than deleting rows indiscriminately. Tables without declared foreign keys gain no relationship checks from this command; an empty result is not complete business validation or a disk-integrity guarantee.

6. Troubleshooting: separate constraint failures from lock contention

For resumable batches, keep result and progress database updates in one transaction. Foreign keys constrain database relationships, not exactly-once execution of external files, HTTP requests, or an entire workflow.

Summary

After declaring foreign keys, enable and verify checks outside a transaction on every connection. Use the same valid-plus-invalid batch to prove rejection and full rollback, then separately audit historical violations. Keep backups and business evidence for repairs instead of treating disabled checks as a fix.