Skip to Content

Ash is a developer

Simple Migration System in SQLite

Updated .

For zsql, I needed a way to set up and migrate the SQLite database that stores history. An ORM would be overkill at the size of my project, so I rolled my own migration system.

Typical ORMs keep migration state in a table. As it turns out, SQLite has a user_version pragma for, well, user versioning. We’ll use it to track the current schema version. It even conveniently defaults to zero!

This post uses Python and its import sqlite3, but language doesn’t matter. It’s all SQLite!

Setup

We define an array-of-arrays of migrations and use its length as the application version:

Python
MIGRATIONS = [
  # Each sub-array here is a migration as a
  # list of SQL statements
  [
    "CREATE TABLE products(name TEXT NOT NULL)",
    "CREATE INDEX products_by_name ON products(name)"
  ],
  # Version 2 of our example schema starts tracking
  # sales.
  [
    "ALTER TABLE products ADD COLUMN sales INT"
  ]
]
APP_VERSION = len(migrations)

And helpers to wrap the version-handling so it doesn’t complicate migrating:

Python
def get_user_version(conn):
  cursor = conn.execute("PRAGMA user_version")
  return cursor.fetchone()[0]

def set_user_version(conn, version):
  # Pragma statements can't be parameterized in
  # SQLite, so this string substitution is necessary.
  conn.execute(f"PRAGMA user_version={version}")

class MigrationException(Exception):
  pass

Migrating

user_version is treated as an array index into MIGRATIONS that can be saved and loaded from the database. We only need to run the migrations at and above the user version we load.

Python
def migrate(conn):
  user_version = get_user_version(conn)

  if user_version < 0 or user_version > APP_VERSION:
    raise MigrationException("unknown version")

  if user_version < APP_VERSION:
    previous_isolation = conn.isolation_level
    try:
      # isolation_level=None to turn off automatic
      # transactions, then begin an exclusive one.
      conn.isolation_level = None
      cursor = conn.cursor()
      try:
        cursor.execute("BEGIN EXCLUSIVE")

        # Slice after user_version to get
        # new migrations.
        for migration in migrations[user_version:]:
          for statement in migration:
            cursor.execute(statement)

        set_user_version(cursor, APP_VERSION)
      # Rollback and re-raise when in error,
      # otherwise commit.
      except:
        cursor.execute("ROLLBACK")
        raise
      else:
        cursor.execute("COMMIT")
      finally:
        cursor.close()
    finally:
      conn.isolation_level = previous_isolation

I’ve found this pattern to be helpful for smaller applications using SQLite. It’s generalizable and doesn’t require much code, so it’s easy to adapt to what’s needed.

Caveats

Migrations are ordered and keyed by array index. Developers who want to merge migrations must keep the migration array ordered by merge date (or another system) to avoid a release which skips migrations or runs them multiple times.

SQLite has limited ability to alter the schema, particularly with alter table. You’ll have to write migrations so they’re compatible with SQLite.