core

Core migrator code

db = database(":memory:")

Migrator class adds a migrations table to your database that allows you to keep track of your database schema changes:

---

source

Migrator

 Migrator (db)

Initialize self. See help(type(self)) for accurate signature.

Migrator(db)
db.t

migrations

assert len(list(db.t)) == 1

---

source

Migrator.add_rollback

 Migrator.add_rollback (rollback_id:int)

---

source

Migrator.add_migration

 Migrator.add_migration (migration_id:int)

---

source

Migrator.migrate

 Migrator.migrate ()

In order to run migrations, you have to make sure all your registered migration ids follow a sequence 0,1,2, …

db = database(":memory:")
m = Migrator(db)

@m.add_migration(0)
def init_db(db): pass
@m.add_migration(1)
def do_something(db): pass
# Error:
@m.add_migration(10)
def do_something(db): pass

test_fail(m.migrate)

Migrations are run inside a transaction, so if something fails in the middle of migration, it won’t apply migration at all:

db = database(":memory:")
m = Migrator(db)

@m.add_migration(0)
def init_db(db):
    db.q("CREATE TABLE cats (id INTEGER PRIMARY KEY AUTOINCREMENT)")
    db.q("CREATE TABLE rats (id INTEGER PRIMARY KEY AUTOINCREMENT)")
    db.q("INSERT INTO cats (id) VALUES (1), (2)")
    db.q("CREATE TABLE bats (id TEXT PRIMARY KEY AUTOINCREMENT)")

@m.add_migration(1)
def wont_run(db): print("first migration fails, so this one won't run")

test_fail(m.migrate)

SQLITE_LOG: AUTOINCREMENT is only allowed on an INTEGER PRIMARY KEY in "CREATE TABLE bats (id TEXT PRIMARY KEY AUTOINCREMENT)" (1) SQLITE_ERROR

0 init_db

db.t, db.t.migrations()

(migrations, [])

assert len(list(db.t)) == 1
assert not db.t.migrations()

---

source

Migrator.last_applied_migration

 Migrator.last_applied_migration ()

---

source

Migrator.applied_migrations

 Migrator.applied_migrations ()

---

source

Migrator.rollback

 Migrator.rollback ()

Rollbacks are also run inside a transaction:

db = database(":memory:")
m = Migrator(db)

@m.add_migration(0)
def init_db(db):
    db.q("CREATE TABLE cats (id INTEGER PRIMARY KEY AUTOINCREMENT)")
    db.q("CREATE TABLE rats (id INTEGER PRIMARY KEY AUTOINCREMENT)")
    db.q("INSERT INTO cats (id) VALUES (1), (2)")
    db.q("CREATE TABLE bats (id INTEGER PRIMARY KEY AUTOINCREMENT)")

@m.add_migration(1)
def will_run(db): print("hello")

@m.add_rollback(1)
def will_not_run(db):
    db.q("DROP TABLE cats")
    db.q("DROP TABLE mats") # mats table doesn't exist, so this will error

m.migrate()
test_fail(m.rollback)

0 init_db
1 will_run
hello
1 will_not_run

assert 'cats' in db.t

Add migrations by decorating your functions like so:

db = database(":memory:")
m = Migrator(db)

@m.add_migration(0)
def init_db(db): db.q("CREATE TABLE cats (name PRIMARY KEY)")

@m.add_migration(1)
def add_dogs(db): db.q("CREATE TABLE dogs (name PRIMARY KEY)")

Running m.migrate would apply these migrations in order of migrations_id.

m.migrate()

0 init_db
1 add_dogs

db.t

cats, dogs, migrations

Running it again does nothing:

m.migrate()

What if you realize there is something wrong with the last migration? You can write a rollback function to fix it! Make sure rollback_id matches corresponding migration_id

@m.add_rollback(1)
def remove_dogs(db): db.q("DROP TABLE dogs")

m.applied_migrations()

[Migrations(id=0, name='init_db', inserted_at='2025-03-09 19:23:08'),
 Migrations(id=1, name='add_dogs', inserted_at='2025-03-09 19:23:08')]

m.last_applied_migration()

Migrations(id=1, name='add_dogs', inserted_at='2025-03-09 19:23:08')

m.rollback()

1 remove_dogs

m.applied_migrations()

[Migrations(id=0, name='init_db', inserted_at='2025-03-09 19:23:08')]

db.t

cats, migrations

assert len(m.applied_migrations()) == 1
assert len(list(db.t)) == 2

Patch database directly

For simpler API, let’s patch add everything directly to database!

---

source

database

 database (path, wal=True)

db = database(":memory:")

@db.add_migration(0)
def init_db(db): db.q("CREATE TABLE cats (name PRIMARY KEY)")

db.migrate()

0 init_db

db.rollback_migration()

No rollback for the latest applied migration found: Migrations(id=0, name='init_db', inserted_at='2025-03-09 19:23:08')