37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from .db import db
|
|
|
|
# format: [str|tuple(str, any...)|callable]
|
|
MIGRATIONS = [
|
|
|
|
]
|
|
|
|
def run_migrations():
|
|
db.execute("""
|
|
CREATE TABLE IF NOT EXISTS _migrations(
|
|
id INTEGER PRIMARY KEY
|
|
)
|
|
""")
|
|
if len(MIGRATIONS) == 0:
|
|
print("No migrations defined.")
|
|
return
|
|
print("Running migrations...")
|
|
ran = 0
|
|
completed = {int(row["id"]) for row in db.query("SELECT id FROM _migrations")}
|
|
to_run = {idx: migration_obj for idx, migration_obj in enumerate(MIGRATIONS) if idx not in completed}
|
|
if not to_run:
|
|
print('No migrations need to run.')
|
|
return
|
|
|
|
with db.transaction():
|
|
for migration_id, migration_obj in to_run.items():
|
|
if isinstance(migration_obj, str):
|
|
db.execute(migration_obj)
|
|
elif isinstance(migration_obj, tuple):
|
|
db.execute(migration_obj[0], *migration_obj[1:])
|
|
elif callable(migration_obj):
|
|
migration_obj()
|
|
|
|
db.execute('INSERT INTO _migrations (id) VALUES (?)', migration_id)
|
|
ran += 1
|
|
print(f"Ran {ran} migrations.")
|