Migrating a CLI’s SQLite Schema Between Releases

A server application migrates one database, on a deploy you control. A CLI migrates thousands of databases — one on every user's machine — whenever each user happens to upgrade, from whatever version they happen to be on. Some will jump from 1.2 straight to 3.0. Some will run 3.0 once, then go back to 2.9 because of a bug. Some will press Ctrl-C halfway through the first run of the new version. A migration system for a CLI has to handle all of that without anyone watching, and without losing the history or state the user has accumulated. This guide builds one on SQLite's user_version pragma, with SQL and Python migration steps, the table-rebuild technique for changes ALTER TABLE cannot make, a guard against downgrades, automatic backups and a test for every historical version. It belongs to the SQLite state topic.

Prerequisites

• Python 3.10+ and a CLI that already stores state in SQLite, for example with the repository class from storing CLI state in SQLite.
• Familiarity with SQL CREATE TABLE, ALTER TABLE and transactions.

The model: an ordered list and one integer

Every SQLite database has a 32-bit integer in its header, user_version, which SQLite itself never touches. It starts at 0. The migration system treats it as "the number of migrations applied", keeps an ordered list of migrations in the code, and on every start applies the ones the database has not seen yet.

The rules that make this robust are few. Migrations are append-only: once released, a migration is never edited or reordered, because some user's database already ran it. Each migration runs in its own transaction together with the version bump, so the database is always at exactly one known version. And the code refuses to touch a database newer than it understands, because an old release cannot know what a newer one changed.

The recipe

# src/mytool/migrations.py
from __future__ import annotations

import sqlite3
from collections.abc import Callable
from contextlib import closing
from pathlib import Path

Step = str | Callable[[sqlite3.Connection], None]


def _split_tags(conn: sqlite3.Connection) -> None:
    """v3: move comma-separated tags into their own table."""
    rows = conn.execute("SELECT id, tags FROM notes WHERE tags != ''").fetchall()
    conn.executemany(
        "INSERT INTO note_tags (note_id, tag) VALUES (?, ?)",
        [(note_id, t.strip()) for note_id, tags in rows for t in tags.split(",") if t.strip()],
    )


def _rebuild_notes_without_tags(conn: sqlite3.Connection) -> None:
    """v4: drop the old tags column by rebuilding the table."""
    conn.execute("""CREATE TABLE notes_new (
                        id INTEGER PRIMARY KEY,
                        text TEXT NOT NULL,
                        created_at TEXT NOT NULL)""")
    conn.execute("INSERT INTO notes_new (id, text, created_at) "
                 "SELECT id, text, created_at FROM notes")
    conn.execute("DROP TABLE notes")
    conn.execute("ALTER TABLE notes_new RENAME TO notes")


MIGRATIONS: list[Step] = [
    # v1
    "CREATE TABLE notes (id INTEGER PRIMARY KEY, text TEXT NOT NULL, tags TEXT NOT NULL DEFAULT '')",
    # v2
    "ALTER TABLE notes ADD COLUMN created_at TEXT NOT NULL DEFAULT '1970-01-01T00:00:00+00:00'",
    # v3: two statements and a data migration, applied together
    lambda conn: (
        conn.execute("CREATE TABLE note_tags (note_id INTEGER NOT NULL REFERENCES notes(id) "
                     "ON DELETE CASCADE, tag TEXT NOT NULL, PRIMARY KEY (note_id, tag))"),
        _split_tags(conn),
    ),
    # v4
    _rebuild_notes_without_tags,
]
LATEST = len(MIGRATIONS)


class SchemaTooNew(Exception):
    def __init__(self, found: int) -> None:
        super().__init__(f"database schema v{found} is newer than this version of mytool "
                         f"understands (v{LATEST}); upgrade mytool to use it")
        self.found = found


def current_version(conn: sqlite3.Connection) -> int:
    return conn.execute("PRAGMA user_version").fetchone()[0]


def backup(path: Path, version: int) -> Path:
    dest = path.with_name(f"{path.name}.v{version}.bak")
    with closing(sqlite3.connect(path)) as src, closing(sqlite3.connect(dest)) as dst:
        src.backup(dst)
    return dest


def migrate(conn: sqlite3.Connection, *, db_path: Path | None = None) -> list[int]:
    """Apply pending migrations; return the versions applied."""
    found = current_version(conn)
    if found > LATEST:
        raise SchemaTooNew(found)
    if found == LATEST:
        return []
    if db_path is not None and found > 0:
        backup(db_path, found)                       # only for existing data
    applied = []
    previous = conn.isolation_level
    conn.isolation_level = None                      # we issue BEGIN/COMMIT ourselves
    conn.execute("PRAGMA foreign_keys = OFF")        # needed for rebuilds; no-op inside BEGIN
    try:
        for number in range(found + 1, LATEST + 1):
            step = MIGRATIONS[number - 1]
            conn.execute("BEGIN IMMEDIATE")          # step + version bump: one transaction
            try:
                if isinstance(step, str):
                    conn.execute(step)
                else:
                    step(conn)
                conn.execute(f"PRAGMA user_version = {number}")
                conn.execute("COMMIT")
            except BaseException:
                conn.execute("ROLLBACK")
                raise
            applied.append(number)
        problems = conn.execute("PRAGMA foreign_key_check").fetchall()
        if problems:
            raise sqlite3.IntegrityError(f"foreign key violations after migration: {problems[:3]}")
    finally:
        conn.execute("PRAGMA foreign_keys = ON")
        conn.isolation_level = previous
    return applied

Why the explicit BEGIN

It is tempting to write each step as with conn: conn.execute(step), and it looks transactional. It is not, for schema changes. In its default (legacy) transaction mode, Python's sqlite3 module only opens a transaction implicitly before INSERT, UPDATE, DELETE and REPLACE — not before CREATE, ALTER, DROP or PRAGMA. A step that creates a table and then fails, or a crash between the CREATE TABLE and the user_version bump, leaves the table behind while the version still says it was never created; the next run then dies with "table already exists". Switching the connection to isolation_level = None and issuing BEGIN IMMEDIATE, COMMIT and ROLLBACK yourself makes every step genuinely all-or-nothing, because SQLite itself supports transactional DDL. IMMEDIATE takes the write lock up front, so two runs migrating at once queue cleanly instead of deadlocking on a lock upgrade. (On Python 3.12+, opening the connection with autocommit=False gives the same effect with explicit commit() and rollback() calls.) The failed-step test below exists precisely to catch this.

Three kinds of step

Plain SQL strings cover most changes: new tables, new indexes, and ALTER TABLE ... ADD COLUMN. SQLite's ALTER TABLE can add, rename and (since 3.35) drop simple columns, but it cannot change a column's type, add a constraint, or drop a column that is indexed or referenced.

Python callables handle data migrations — splitting a column into a table, normalising values, computing a new field — where SQL alone is awkward. They receive the connection inside the transaction, so a failure rolls back both the data changes and the schema change.

Table rebuilds handle everything ALTER TABLE cannot: create the new table with the desired definition, copy the data across, drop the old table, rename the new one. SQLite's documentation describes this procedure, and it is safe inside a transaction. Foreign key enforcement must be off during the rebuild — otherwise dropping the old table would cascade or fail — which is why migrate() turns it off for the duration and runs PRAGMA foreign_key_check afterwards to prove the result is consistent.

UX considerations

• Migrate automatically, silently when fast. Users should not have to run a migrate command; opening the database migrates it. Print a one-line note on stderr only when a migration takes noticeable time or makes a backup.
• Back up before changing existing data. The .v3.bak file next to the database costs a little disk and saves the day when a migration has a bug. Keep the last one or two, and mention them in the release notes.
• Explain downgrades clearly. SchemaTooNew turns "no such column" crashes into "this database was created by a newer mytool; upgrade, or run mytool db reset to start over". Friendly error messages and tracebacks shows where to catch it.
• Avoid long migrations on hot paths. If a data migration might take minutes on a large database, say so before starting and show progress, as in adding progress bars and spinners.
• Caches do not need migrations. If a cache's schema changes, delete and recreate it. Reserve migrations for data worth keeping.

Testing the behaviour

The test that matters most builds a database at every historical version with realistic data, migrates it, and checks the result. Because migrations are append-only, a database at version N can be built by applying the first N steps:

# tests/test_migrations.py
import sqlite3

import pytest

from mytool import migrations as m


def at_version(path, version):
    conn = sqlite3.connect(path)
    for number, step in enumerate(m.MIGRATIONS[:version], start=1):
        with conn:
            conn.execute(step) if isinstance(step, str) else step(conn)
            conn.execute(f"PRAGMA user_version = {number}")
    return conn


def seed_v1(conn):
    with conn:
        conn.execute("INSERT INTO notes (id, text, tags) VALUES (1, 'milk', 'home, errand')")
        conn.execute("INSERT INTO notes (id, text, tags) VALUES (2, 'call Bo', '')")


@pytest.mark.parametrize("start", range(0, m.LATEST + 1))
def test_every_version_reaches_latest(tmp_path, start):
    conn = at_version(tmp_path / "db.sqlite", start)
    m.migrate(conn)
    assert m.current_version(conn) == m.LATEST
    cols = [r[1] for r in conn.execute("PRAGMA table_info(notes)")]
    assert cols == ["id", "text", "created_at"]


def test_data_survives_from_v1(tmp_path):
    path = tmp_path / "db.sqlite"
    conn = at_version(path, 1)
    seed_v1(conn)
    applied = m.migrate(conn, db_path=path)
    assert applied == [2, 3, 4]
    tags = conn.execute("SELECT tag FROM note_tags WHERE note_id = 1 ORDER BY tag").fetchall()
    assert [t[0] for t in tags] == ["errand", "home"]
    assert (path.parent / "db.sqlite.v1.bak").exists()


def test_failed_step_rolls_back(tmp_path, monkeypatch):
    conn = at_version(tmp_path / "db.sqlite", 2)
    def boom(conn):
        conn.execute("CREATE TABLE half_done (x)")
        raise RuntimeError("bug in migration")
    monkeypatch.setattr(m, "MIGRATIONS", [*m.MIGRATIONS[:2], boom, *m.MIGRATIONS[3:]])
    with pytest.raises(RuntimeError):
        m.migrate(conn)
    assert m.current_version(conn) == 2
    assert conn.execute("SELECT name FROM sqlite_master WHERE name = 'half_done'").fetchone() is None


def test_newer_database_is_refused(tmp_path):
    conn = sqlite3.connect(tmp_path / "db.sqlite")
    conn.execute(f"PRAGMA user_version = {m.LATEST + 1}")
    with pytest.raises(m.SchemaTooNew):
        m.migrate(conn)

The parametrised test grows automatically with every new migration, so a release can never ship a step that breaks upgrades from an older version. For extra confidence, keep a few real database files from old releases as fixtures — they catch data shapes that synthetic seeds miss.

Conclusion

A CLI's migrations run unattended on every user's machine, so they must be ordered, append-only, transactional and defensive. PRAGMA user_version gives you the version counter for free; a list of SQL strings and Python callables gives you the steps; table rebuilds cover what ALTER TABLE cannot; a backup and a downgrade guard cover the cases where things go wrong. Test every historical version on every change, and schema evolution stops being the scariest part of a release.

Frequently asked questions

Related

• Up: Local state and SQLite in Python CLIs
• Storing CLI state in SQLite
• Recording and querying CLI run history
• Semantic versioning policy for CLI tools
• Versioning a plugin API