suckless SQLite schema migrations in python

It happened again. I wrote something that started with a small CREATE TABLE IF NOT EXISTS statement, that kept growing while manually executing DROP statements and changing it as needed. At some point I knew I would need some kind of migration system.

I’ve been there before. I searched “sql migration tool”. The results where the same as ever, a mixed bag of either declarative schema migration tools, alembic and yet more tools that are alpha, beta and carry an ORM by their hand. And that’s fine, I guess.

It happens that having been there before you know that sunk cost applies to the tool you end up going with. And then you go: how hard really is this, why not build my own. But then again, I remember all the features migration tools give me: going back between migrations, cli, etc. But then again, I also remember the troubles glueing together these tools with test suites, how most of the times I end up writing plain SQL, and so on and so forth.

This time I got lucky and found this blog post that gave me a pass for rolling my own, using PRAGMA user_version to keep track of the current schema.

And this time, I decided to go with it. This is my version, and I will go with this until it breaks and stops doing its thing. It is alpha. But at least, it’s my alpha.

First, there’s a folder with migrations.

$ ls migrations/
0001_init.sql
0002_add_foo.sql

Each migration starts setting PRAGMA user_version to a number.

$ head -n 1 migrations/0001_init.sql
PRAGMA user_version=1;
$ head -n 1 migrations/0002_add_foo.sql
PRAGMA user_version=2;

And then, there’s this I run somewhere.

import os
import sys
import sqlite3
import logging

log = logging.getlogger("foo")

DB_URI = os.getenv("DB_URI", "foo.db")
conn = sqlite3.connect(db_uri)

current_version, = next(conn.cursor().execute('PRAGMA user_version'), (0, ))

migrations = resources.files('migrations').iterdir()

for migration in migrations[current_version:]:
    cur = conn.cursor()
    try:
        log.info("Applying %s", migration.name)
        cur.executescript("begin;" + migration.read_text())
    except Exception as e:
        log.error("Failed migration %s: %s. Bye", migration.name, e)
        cur.execute("rollback")
        sys.exit(1)
    else:
        cur.execute("commit")

And then I am done. Happy that I do not need to spend more time deciding between tools and if ORMs are my thing or not.