Building a SQLite Migration Runner in Python

#frontend #webdev

Building a SQLite Migration Runner in Python

A small migration runner is a practical tool every team with embedded databases eventually needs. This tutorial shows how to version schema changes, apply them safely, and keep a simple audit trail using plain SQL and Python.

Why this pattern works

SQLite exposes a built-in version slot called PRAGMA user_version, which you can use to track the current schema version. The core idea is simple: read the version on startup, run every numbered migration you have not applied yet, and bump the version only after each migration succeeds.

This approach is useful because it keeps the database self-describing without needing a heavy framework. It also makes local development, CI, and deployed databases converge on the same schema state in a predictable way.

Project structure

A clean layout keeps migrations easy to review and hard to break. Use one file per migration, number them in order, and store them in a dedicated folder.

project/
  app.py
  migrations/
    001_initial.sql
    002_add_created_at.sql
    003_add_indexes.sql

Each migration should do one logical change, because that makes review and rollback simpler. Once a migration is shipped, do not edit it; add a new one instead.

The migration rules

Follow four rules from the start. First, migrations are numbered consecutively. Second, each migration runs inside a transaction. Third, user_version changes only after the schema change succeeds. Fourth, never rewrite a shipped migration.

These rules prevent the classic “works on my machine” problem where different machines have different schema history. They also make failures safer because SQLite can roll back transactional schema changes cleanly.

Write your SQL files

Start with a migration that creates your initial table. Keep the SQL explicit and readable so future changes stay obvious.

 migrations/001_initial.sql
BEGIN;

CREATE TABLE IF NOT EXISTS notes (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  title TEXT NOT NULL,
  body TEXT NOT NULL
);

PRAGMA user_version = 1;

COMMIT;

Then add a second migration for a common 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;

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()

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}")

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;

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

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:

Create migrations/001_initial.sql with the base table.
Create migrations/002_add_created_at.sql for the timestamp column.
Add app.py with the Python runner.
Run python app.py on a new database and confirm user_version = 2.
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