Resume Python Batch Jobs with SQLite Checkpoints

A batch script finishes hundreds of records, stops, and starts from the beginning on its next run. A small SQLite database can keep the finished results so the next run has less work to do.

This tutorial builds a single-worker batch runner. Each completed result becomes one committed row. On restart, the script checks the current inputs and skips matching rows. The demo counts text characters, so you can exercise the recovery path without a model, API key, or external tool.

1. Keep the result and completion record together

A checkpoint that says done=true while the output is missing creates a recovery problem. Here, the stored result is also the completion record:

check input fingerprint → calculate → serialize result → insert row → commit → report DONE

The table has three columns: a stable job ID, a fingerprint of the input and worker version, and the JSON result. The job ID is the primary key. Reusing an ID with different text is an error; changing the worker version also invalidates that ID’s previous result.

This example assumes one process owns the database on a suitable local filesystem. It keeps results inside SQLite. Large output files, concurrent workers, and API side effects need additional coordination.

2. Download and pause a small batch

Use Python 3.12 or newer with the standard sqlite3 module available. The examples were tested on macOS with Python 3.14.7. Shell examples assume bash or zsh.

Download resume_batch.py into a new working directory. Save this as jobs.json:

[
  {"id": "seq_001", "text": "MKT"},
  {"id": "seq_002", "text": "ACDE"},
  {"id": "seq_003", "text": "GGGGG"}
]

Run:

python3 --version
python3 resume_batch.py --help
mkdir sqlite-resume-demo
python3 resume_batch.py jobs.json sqlite-resume-demo/progress.sqlite --stop-after 2

Expected output from the batch command:

DONE seq_001
DONE seq_002
STOPPED: new=2; skipped=0

--stop-after pauses after that many newly committed results and returns exit status 0. Skipped rows do not count toward the limit. This is a repeatable pause demonstration; it does not simulate a power failure. Use a fresh demo directory when repeating the full walkthrough.

3. Resume, then rerun again

Run the same input with the same database, without the pause option:

python3 resume_batch.py jobs.json sqlite-resume-demo/progress.sqlite

The first two results are already present:

SKIP seq_001
SKIP seq_002
DONE seq_003
FINISHED: new=1; skipped=2

Run that command once more. It prints three SKIP lines and FINISHED: new=0; skipped=3.

DONE is printed after the database commit. A process could stop between committing and printing, so a missing terminal message does not prove that the result is missing. Restart decisions come from the database.

4. The complete implementation

"""Resume a single-worker text batch using SQLite. Python 3.12+."""
import argparse
import hashlib
import json
from pathlib import Path
import sqlite3
import sys

WORKER_VERSION = "text-length-v1"


def load_jobs(path):
    jobs = json.loads(Path(path).read_text(encoding="utf-8"))
    if not isinstance(jobs, list):
        raise ValueError("input must be a JSON list")
    seen = set()
    for job in jobs:
        if not isinstance(job, dict) or set(job) != {"id", "text"}:
            raise ValueError("each job needs exactly id and text")
        job_id, text = job["id"], job["text"]
        if not isinstance(job_id, str) or not job_id.strip():
            raise ValueError("job id must be a nonempty string")
        if not isinstance(text, str):
            raise ValueError("job text must be a string")
        if job_id in seen:
            raise ValueError(f"duplicate job id: {job_id}")
        seen.add(job_id)
    return jobs


def fingerprint(text):
    payload = json.dumps([WORKER_VERSION, text], ensure_ascii=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def calculate(text):
    return {"length": len(text)}


def run_batch(jobs, database, *, stop_after=None):
    completed = skipped = 0
    con = sqlite3.connect(database, autocommit=False)
    try:
        with con:
            con.execute("""CREATE TABLE IF NOT EXISTS results (
                job_id TEXT PRIMARY KEY,
                fingerprint TEXT NOT NULL,
                result_json TEXT NOT NULL
            )""")
        # Check every current input before calculating any new result.
        for job in jobs:
            old = con.execute(
                "SELECT fingerprint FROM results WHERE job_id = ?",
                (job["id"],),
            ).fetchone()
            if old is not None and old[0] != fingerprint(job["text"]):
                raise ValueError(f"input or worker changed for {job['id']}; use a new database")
        for job in jobs:
            old = con.execute(
                "SELECT 1 FROM results WHERE job_id = ?", (job["id"],)
            ).fetchone()
            if old is not None:
                skipped += 1
                print(f"SKIP {job['id']}", flush=True)
                continue
            result = calculate(job["text"])
            encoded = json.dumps(result, allow_nan=False, sort_keys=True)
            # The result row itself is the completion record.
            with con:
                con.execute(
                    "INSERT INTO results VALUES (?, ?, ?)",
                    (job["id"], fingerprint(job["text"]), encoded),
                )
            completed += 1
            print(f"DONE {job['id']}", flush=True)
            if stop_after is not None and completed >= stop_after:
                print(f"STOPPED: new={completed}; skipped={skipped}", flush=True)
                return
        print(f"FINISHED: new={completed}; skipped={skipped}", flush=True)
    finally:
        con.close()


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path, help="JSON list of id/text jobs")
    parser.add_argument("database", type=Path, help="SQLite checkpoint; parent must exist")
    parser.add_argument("--stop-after", type=positive_int, help="pause after N new results")
    args = parser.parse_args()
    try:
        jobs = load_jobs(args.input)
        run_batch(jobs, args.database, stop_after=args.stop_after)
    except (OSError, ValueError, sqlite3.Error) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("INTERRUPTED: rerun with the same input and database", file=sys.stderr)
        return 130
    return 0


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

The CLI validates the whole JSON list before opening the database. IDs must be nonempty strings and unique; text must be a string. Empty text is valid and has length zero. SQL values use placeholders rather than string interpolation.

The fingerprint includes WORKER_VERSION and the exact text. Input order can change. An existing ID’s text cannot change without selecting a new database. The fingerprint does not discover changes to Python code automatically: update the version whenever calculation behavior changes. Include model identifiers, parameters, and relevant dependencies in the identity for a real inference pipeline.

The code selects autocommit=False explicitly. Python’s transaction documentation explains this mode. The connection context manager commits a successful block and rolls back a failed one; it does not close the connection, which is why the code has finally: con.close().

5. Inspect the stored results

Save this as inspect_results.py beside the downloaded script:

import json
import sqlite3

con = sqlite3.connect("sqlite-resume-demo/progress.sqlite")
try:
    for job_id, result_json in con.execute(
        "SELECT job_id, result_json FROM results ORDER BY job_id"
    ):
        print(job_id, json.loads(result_json)["length"])
finally:
    con.close()

Run python3 inspect_results.py:

seq_001 3
seq_002 4
seq_003 5

The demo counts Python string characters, not UTF-8 bytes or visible grapheme clusters. It does not validate protein sequences. Replace calculate() with your own deterministic computation, validate its result, and update WORKER_VERSION.

The database retains jobs removed from the input list. This inspection query prints all saved rows; filter by the current job IDs when exporting results for a particular input file.

6. Check that changed input is rejected

Save this as changed.json:

[
  {"id": "seq_001", "text": "MKA"},
  {"id": "seq_002", "text": "ACDE"},
  {"id": "seq_003", "text": "GGGGG"}
]

Run:

python3 resume_batch.py changed.json sqlite-resume-demo/progress.sqlite

It returns exit status 1 and reports:

ERROR: input or worker changed for seq_001; use a new database

Both strings have length three, but their content differs. The check catches the changed input before calculating any new jobs. Existing results remain available. Use a fresh path such as sqlite-resume-demo/changed.sqlite to process the changed dataset.

A dataset checksum manifest can identify the wider collection of files behind your inputs. The per-job fingerprint answers the narrower question of whether a saved row belongs to the current job definition.

7. Understand the recovery boundary

The validation for this article checked pause/resume, repeated skips, same-length input changes, changed worker versions, malformed inputs, Unicode, database insertion failure, and abrupt process exit before the third result was saved. These checks exercised process recovery; they were not power-loss tests.

SQLite’s atomic commit explanation describes transaction recovery and its filesystem and storage assumptions. Keep the database on appropriate storage, preserve its journal files, and avoid weakening SQLite’s durability settings without understanding the consequences.

If your worker calls an API or writes another file, a crash may happen after that action succeeds but before SQLite records it. The next run can repeat the action. Use an idempotency key supported by the service, or design outputs so that retrying the same job is safe. This tutorial does not promise exactly-once external effects.

For an external command, the subprocess logging helper supplies timeouts and logs. Check the generated output before committing a result. For small standalone state documents, atomic JSON writes remain a useful companion pattern.

8. Troubleshooting

9. Summary

Start with stable IDs, record the inputs that determine a result, and commit only after the result is ready. A restart then becomes a check of saved work followed by the remaining calculations.