Schema Migrations and Versioning

Treat the schema like code

Production SQLite schemas evolve over time. The two questions you must answer up-front:

How does a fresh install bring the database to the current version?
How does an existing install upgrade safely from version N to N+1?

SQLite ships PRAGMA user_version — a 32-bit integer stored in the database header. Use it as your migration cursor. The pattern (used by Pippa's backend/store/conversations.py and a thousand others):

On startup, read PRAGMA user_version.
Apply each migration step from current to target inside a transaction.
Bump user_version as the last statement of each migration.

Self-reference: Pippa's own SQLite store uses exactly this pattern — see _init_db() in backend/store/conversations.py. The user_version increments every time we add a column or table; healing logic on every conversation GET ensures the schema and the JSONL ground truth stay aligned.

Code

Migration runner — pure SQL·sql

PRAGMA user_version;
-- 0

BEGIN;
CREATE TABLE conversations (
  id INTEGER PRIMARY KEY,
  title TEXT NOT NULL,
  created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
PRAGMA user_version = 1;
COMMIT;

-- Later, version 2 adds a column
BEGIN;
ALTER TABLE conversations ADD COLUMN brain TEXT NOT NULL DEFAULT 'claude';
PRAGMA user_version = 2;
COMMIT;

Migration runner — Python·python

import sqlite3

MIGRATIONS = [
    """CREATE TABLE conversations (
         id INTEGER PRIMARY KEY,
         title TEXT NOT NULL,
         created_at TEXT NOT NULL DEFAULT (datetime('now'))
       )""",
    "ALTER TABLE conversations ADD COLUMN brain TEXT NOT NULL DEFAULT 'claude'",
    "CREATE INDEX idx_conv_brain_created ON conversations(brain, created_at DESC)",
]

def migrate(db_path: str) -> int:
    with sqlite3.connect(db_path) as conn:
        current = conn.execute('PRAGMA user_version').fetchone()[0]
        for i, sql in enumerate(MIGRATIONS, start=1):
            if current >= i:
                continue
            conn.execute('BEGIN')
            try:
                conn.executescript(sql)
                conn.execute(f'PRAGMA user_version = {i}')
                conn.commit()
            except Exception:
                conn.rollback()
                raise
        return conn.execute('PRAGMA user_version').fetchone()[0]

print('migrated to version', migrate('myapp.db'))

Exercise

Build a tiny migration runner: a Python function that takes a list of SQL strings (each a 'migration step') and applies them in order, bumping PRAGMA user_version after each. Test with at least three migrations on a fresh database and on a database already at version 2 — the runner should skip what's done and apply only what's new. Bonus: wrap each migration in a transaction with rollback on failure.

Schema Migrations and Versioning

Treat the schema like code

Code

External links

Exercise

Progress

Comments 0