Building a SQLite Migration Runner in Python

schema evolution, such as a timestamp column. SQLite supports ADD COLUMN, so small additions like this are straightforward.

 migrations/002_add_created_at.sql
BEGIN;

ALTER TABLE notes ADD COLUMN created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP;

PRAGMA user_version = 2;

COMMIT;

Enter fullscreen mode Exit fullscreen mode

If you need a bigger change that SQLite cannot express directly, rebuild the table instead of forcing an unsupported ALTER TABLE. SQLite’s built-in ALTER TABLE support is intentionally limited.

Build the runner

The Python code below reads the current version, finds pending migration files, and applies them in order. It also records each successful migration in a small audit table so you can see what ran and when.

from pathlib import Path
import sqlite3
from datetime import datetime

MIGRATIONS_DIR = Path("migrations")

def get_migrations():
    files = sorted(MIGRATIONS_DIR.glob("*.sql"))
    migrations = []
    for path in files:
        version = int(path.stem.split("_", 1))
        migrations.append((version, path))
    return migrations

def ensure_audit_table(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS schema_migrations (
            version INTEGER PRIMARY KEY,
            filename TEXT NOT NULL,
            applied_at TEXT NOT NULL
        )
    """)

def current_version(conn):
    row = conn.execute("PRAGMA user_version").fetchone()
    return row

def apply_migration(conn, version, path):
    sql = path.read_text(encoding="utf-8")
    with conn:
        conn.executescript(sql)
        conn.execute(
            "INSERT OR REPLACE INTO schema_migrations(version, filename, applied_at) VALUES (?, ?, ?)",
            (version, path.name, datetime.utcnow().isoformat(timespec="seconds") + "Z")
        )
        conn.execute(f"PRAGMA user_version = {version}")

def migrate(db_path="app.db"):
    conn = sqlite3.connect(db_path)
    try:
        ensure_audit_table(conn)
        version = current_version(conn)
        for target_version, path in get_migrations():
            if target_version > version:
                apply_migration(conn, target_version, path)
                version = target_version
    finally:
        conn.close()

if __name__ == "__main__":
    migrate()

Enter fullscreen mode Exit fullscreen mode

Notice that each migration runs in a transaction. If anything fails, the changes are rolled back together, which keeps the database from landing in a half-applied state.

Add safety checks

A good runner should fail fast if your migration numbers are broken. For example, if you accidentally skip from 002 to 004, the code should complain before touching the database.

def validate_sequence(migrations):
    versions = [v for v, _ in migrations]
    expected = list(range(1, len(versions) + 1))
    if versions != expected:
        raise ValueError(f"Expected consecutive versions {expected}, got {versions}")

Enter fullscreen mode Exit fullscreen mode

You can also verify the database version after the loop. That gives you an immediate signal if a migration file was missing, misnamed, or partially copied during deployment.

Handle table rebuilds

Some schema changes require more than ADD COLUMN. SQLite cannot arbitrarily change column types or constraints, so the safest pattern is to create a new table, copy the data, drop the old one, and rename the new table into place.

Here is an example migration that adds a status column by rebuilding the table:

 migrations/003_rebuild_notes.sql
BEGIN;

CREATE TABLE notes_new (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  status TEXT NOT NULL DEFAULT 'draft'
);

INSERT INTO notes_new (id, title, body, created_at)
SELECT id, title, body, created_at
FROM notes;

DROP TABLE notes;

ALTER TABLE notes_new RENAME TO notes;

PRAGMA user_version = 3;

COMMIT;

Enter fullscreen mode Exit fullscreen mode

When rebuilding tables with foreign keys, temporarily disable foreign key checks during the migration and re-enable them afterward. That prevents a rebuild from failing midway because of self-references or dependent tables.

Run it on startup

The runner is most useful when you call it before your app starts serving traffic. That way, your application always uses the latest schema before it handles requests.

def startup():
    migrate("app.db")
    conn = sqlite3.connect("app.db")
    row = conn.execute("SELECT id, title, body, created_at FROM notes").fetchall()
    return row

Enter fullscreen mode Exit fullscreen mode

For small projects, this is often enough. For larger teams, a dedicated migration tool can add locking, retries, and stronger workflow conventions, but the underlying pattern stays the same.

Test it properly

Always test migrations against a copy of real data, not just a fresh empty database. Schema changes can behave differently when the table already has rows, indexes, or foreign keys.

A simple test script can create a temporary database, run the migration runner, and assert that PRAGMA user_version matches the latest migration number. You should also verify that existing rows survive a table rebuild and that the audit table contains one row per applied migration.

A complete example

Here is a compact end-to-end flow for a notes app:

1. Create migrations/001_initial.sql with the base table.
2. Create migrations/002_add_created_at.sql for the timestamp column.
3. Add app.py with the Python runner.
4. Run python app.py on a new database and confirm user_version = 2.
5. Insert a row and verify that the schema and data both work as expected.

This workflow scales surprisingly well because it stays boring in the best possible way. Small, ordered SQL files plus a short runner are enough for many production SQLite apps.

Common mistakes

The biggest mistake is editing old migrations after they have been shared. That creates divergent histories and makes two machines disagree about what “version 2” means.

Another common mistake is making one migration do too much. Keep changes narrow so the next person can understand, test, and revert them without guessing.

A final mistake is forgetting that SQLite has limited ALTER TABLE support. When the change is more than a simple column addition or rename, rebuild the table instead of fighting the engine.

When to use a tool

If your app is tiny, this hand-rolled runner is often the right choice. It is easy to audit, easy to debug, and only depends on the Python standard library and SQLite itself.

If your project grows, a migration framework can save time by handling concurrency, ordered execution, and team workflows. Even then, the mental model remains the same: numbered migrations, transactions, and a version source of truth.

-

Rizwan Saleem | https://rizwansaleem.co