Storing CLI State in SQLite with a Small Repository Class

Incremental tools need memory. A command that uploads a folder should skip files it already uploaded; an importer should skip records it already imported; a thumbnail generator should only touch images that changed. The first implementation is nearly always a JSON file mapping paths to timestamps, loaded at start and written at the end. It works until the folder holds fifty thousand files and every run rewrites a ten-megabyte JSON document, or until a crash halfway through loses the record of everything done so far, or until two runs overlap. This guide replaces that file with SQLite behind a small repository class, so commands ask meaningful questions — "which of these files changed?" — instead of juggling dictionaries. It builds on the connection settings from the SQLite state topic.

Prerequisites

• Python 3.10+; sqlite3 is in the standard library. Typer for the command layer, platformdirs for the location.
• A command that processes items and should skip the unchanged ones on the next run.
• Optionally, a look at caching expensive work between CLI runs, which covers the same idea for computed results.

The design

Commands should not write SQL. A repository class owns the connection, the schema and every query, and exposes methods named after what the command needs. That keeps SQL in one file, makes it easy to test, and means the storage could change later without touching commands.

For an incremental sync, the state is one table: each processed file with the fingerprint it had when it was processed. A file needs work if it is not in the table or its fingerprint changed. Using size plus modification time as the fingerprint is cheap and catches almost every real change; a content hash is more reliable and costs a full read of every file, so offer it as an option.

The recipe

# src/syncer/store.py
from __future__ import annotations

import os
import sqlite3
from collections.abc import Iterable, Iterator
from contextlib import closing, contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from platformdirs import user_state_path

SCHEMA = [
    """CREATE TABLE files (
           path TEXT PRIMARY KEY,
           size INTEGER NOT NULL,
           mtime_ns INTEGER NOT NULL,
           processed_at TEXT NOT NULL
       )""",
    "CREATE INDEX files_processed ON files (processed_at)",
]


@dataclass(frozen=True)
class Fingerprint:
    path: str
    size: int
    mtime_ns: int

    @classmethod
    def of(cls, path: Path) -> Fingerprint:
        st = path.stat()
        return cls(str(path.resolve()), st.st_size, st.st_mtime_ns)


def default_path() -> Path:
    return Path(os.environ.get("SYNCER_DB") or user_state_path("syncer") / "state.db")


class Store:
    def __init__(self, path: Path | None = None) -> None:
        self.path = path or default_path()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(self.path, timeout=10.0)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA journal_mode = WAL")
        self.conn.execute("PRAGMA synchronous = NORMAL")
        self._migrate()

    def _migrate(self) -> None:
        version = self.conn.execute("PRAGMA user_version").fetchone()[0]
        for number, sql in enumerate(SCHEMA[version:], start=version + 1):
            self.conn.execute("BEGIN IMMEDIATE")      # DDL is not auto-wrapped by sqlite3
            try:
                self.conn.execute(sql)
                self.conn.execute(f"PRAGMA user_version = {number}")
                self.conn.execute("COMMIT")
            except BaseException:
                self.conn.execute("ROLLBACK")
                raise

    def close(self) -> None:
        self.conn.close()

    def __enter__(self) -> Store:
        return self

    def __exit__(self, *exc) -> None:
        self.close()

    # -- queries the commands need ------------------------------------------
    def changed(self, prints: Iterable[Fingerprint]) -> list[Fingerprint]:
        """Return the fingerprints that are new or differ from what was recorded."""
        prints = list(prints)
        known = {
            row["path"]: (row["size"], row["mtime_ns"])
            for row in self.conn.execute("SELECT path, size, mtime_ns FROM files")
        }
        return [p for p in prints if known.get(p.path) != (p.size, p.mtime_ns)]

    @contextmanager
    def batch(self) -> Iterator[None]:
        """Group many mark_done calls into one transaction."""
        with self.conn:
            yield

    def mark_done(self, fp: Fingerprint) -> None:
        self.conn.execute(
            """INSERT INTO files (path, size, mtime_ns, processed_at) VALUES (?, ?, ?, ?)
               ON CONFLICT (path) DO UPDATE SET
                   size = excluded.size, mtime_ns = excluded.mtime_ns,
                   processed_at = excluded.processed_at""",
            (fp.path, fp.size, fp.mtime_ns, datetime.now(timezone.utc).isoformat()),
        )

    def forget_missing(self, present: Iterable[str]) -> int:
        """Drop records for files that no longer exist under the synced roots."""
        with self.conn:
            self.conn.execute("CREATE TEMP TABLE IF NOT EXISTS present (path TEXT PRIMARY KEY)")
            self.conn.execute("DELETE FROM present")
            self.conn.executemany("INSERT OR IGNORE INTO present VALUES (?)",
                                  ((p,) for p in present))
            cur = self.conn.execute("DELETE FROM files WHERE path NOT IN (SELECT path FROM present)")
        return cur.rowcount

    def stats(self) -> dict:
        row = self.conn.execute(
            "SELECT COUNT(*) AS n, MAX(processed_at) AS last FROM files").fetchone()
        return {"files": row["n"], "last_processed": row["last"]}

A few details carry most of the value:

• ON CONFLICT ... DO UPDATE is an upsert: one statement inserts a new file or updates an existing one, with no read-then-write race.
• changed() loads the known fingerprints once into a dictionary. For tens of thousands of rows that is faster than one query per file, and still uses a few megabytes at most. For millions, switch to a temporary table and a join, as forget_missing() does.
• batch() groups writes. Committing after every file would mean one disk sync per file; one transaction per batch of a few hundred is dramatically faster and still small enough not to block other runs for long.
• forget_missing() keeps the table from growing forever with records of deleted files — a cleanup that a JSON file usually never gets.

The command

# src/syncer/cli.py
from pathlib import Path

import typer

from syncer.store import Fingerprint, Store

app = typer.Typer()


def upload(path: str) -> None:          # stand-in for the real work
    pass


@app.callback()
def main() -> None:
    """Incremental folder sync."""


@app.command()
def sync(folder: Path, batch_size: int = typer.Option(200, min=1),
         full: bool = typer.Option(False, "--full", help="Ignore recorded state.")) -> None:
    """Upload new and changed files under FOLDER."""
    files = [p for p in folder.rglob("*") if p.is_file()]
    prints = [Fingerprint.of(p) for p in files]
    with Store() as store:
        todo = prints if full else store.changed(prints)
        typer.echo(f"{len(todo)} of {len(prints)} files need syncing", err=True)
        for start in range(0, len(todo), batch_size):
            chunk = todo[start:start + batch_size]
            for fp in chunk:
                upload(fp.path)          # slow work happens outside the transaction
            with store.batch():
                for fp in chunk:
                    store.mark_done(fp)
        removed = store.forget_missing(p.path for p in prints)
        if removed:
            typer.echo(f"forgot {removed} deleted files", err=True)


@app.command()
def status() -> None:
    """Show what the state database knows."""
    with Store() as store:
        s = store.stats()
    typer.echo(f"{s['files']} files recorded, last processed {s['last_processed'] or 'never'}")


if __name__ == "__main__":
    app()

Uploads happen before each chunk's transaction, never inside it, so the write lock is held only for the quick inserts. If the process is interrupted, every completed chunk is already recorded and the next run resumes where this one stopped — the property a write-everything-at-the-end JSON file cannot offer.

UX considerations

• Report the plan before the work. "1,204 of 52,311 files need syncing" tells users the state is working and how long to expect.
• Offer an escape hatch. --full ignores recorded state for one run; a state reset command clears it. Both are what users reach for when they suspect the state is wrong.
• Make the location discoverable. status or --verbose should print the database path, and an environment variable should override it for tests and unusual setups.
• Resume, do not restart. Committing in chunks means Ctrl-C costs at most one chunk of repeated work; mention that in the docs, because users will try it. The signal-handling side is in handling KeyboardInterrupt cleanly.
• Keep paths absolute. Recording resolved paths means running the command from a different directory does not make every file look new.

Testing the behaviour

Use a real database file in tmp_path and real files on disk. Changing a file's modification time with os.utime makes change detection deterministic:

# tests/test_store.py
import os

import pytest

from syncer.store import Fingerprint, Store


@pytest.fixture
def store(tmp_path):
    with Store(tmp_path / "state.db") as s:
        yield s


@pytest.fixture
def files(tmp_path):
    root = tmp_path / "data"
    root.mkdir()
    for name in ("a.txt", "b.txt"):
        (root / name).write_text(name)
    return root


def prints(root):
    return [Fingerprint.of(p) for p in sorted(root.iterdir())]


def test_everything_is_new_at_first(store, files):
    assert len(store.changed(prints(files))) == 2


def test_recorded_files_are_skipped(store, files):
    with store.batch():
        for fp in prints(files):
            store.mark_done(fp)
    assert store.changed(prints(files)) == []


def test_modified_file_is_detected(store, files):
    with store.batch():
        for fp in prints(files):
            store.mark_done(fp)
    target = files / "a.txt"
    st = target.stat()
    os.utime(target, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000_000))
    assert [fp.path for fp in store.changed(prints(files))] == [str(target.resolve())]


def test_deleted_files_are_forgotten(store, files):
    with store.batch():
        for fp in prints(files):
            store.mark_done(fp)
    (files / "b.txt").unlink()
    assert store.forget_missing(fp.path for fp in prints(files)) == 1
    assert store.stats()["files"] == 1


def test_failed_batch_records_nothing(store, files):
    with pytest.raises(RuntimeError):
        with store.batch():
            store.mark_done(prints(files)[0])
            raise RuntimeError("upload failed")
    assert store.stats()["files"] == 0


def test_state_survives_reopening(tmp_path, files):
    with Store(tmp_path / "s.db") as s, s.batch():
        s.mark_done(prints(files)[0])
    with Store(tmp_path / "s.db") as s:
        assert s.stats()["files"] == 1

The failed-batch test documents the transactional guarantee: if anything inside batch() raises, none of that batch's records are kept, so a half-processed chunk is retried on the next run rather than silently marked done.

Conclusion

Moving incremental state from a JSON file to SQLite is a small change with large effects: writes become per-chunk instead of all-or-nothing, interrupted runs resume, queries replace hand-written dictionary logic, and concurrent runs stop overwriting each other. Put the SQL in one repository class with intention-revealing methods, keep slow work outside transactions, and test it against real files and a real database.

Frequently asked questions

Related

• Up: Local state and SQLite in Python CLIs
• Migrating a CLI's SQLite schema
• Recording and querying CLI run history
• File locking for concurrent CLI runs
• Walking directory trees with ignore rules