42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from .db import db
|
|
|
|
def migrate_old_avatars():
|
|
for avatar in db.query('SELECT id, file_path FROM avatars WHERE file_path LIKE "/avatars/%"'):
|
|
new_path = f"/static{avatar['file_path']}"
|
|
db.execute('UPDATE avatars SET file_path = ? WHERE id = ?', new_path, avatar['id'])
|
|
|
|
# format: [str|tuple(str, any...)|callable]
|
|
MIGRATIONS = [
|
|
migrate_old_avatars,
|
|
]
|
|
|
|
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.")
|