move migrations to python scripts

database/migrations.py

···
 import sqlite3
 from pathlib import Path
+from typing import Callable
 
 from database.connection import get_conn
 from util.util import LOGGER
-
 
 class DatabaseMigrator:
     def __init__(self, db_path: Path, migrations_folder: Path) -> None:
         self.db_path: Path = db_path
         self.migrations_folder: Path = migrations_folder
         self.conn: sqlite3.Connection = get_conn(db_path)
+        _ = self.conn.execute("PRAGMA foreign_keys = OFF;")
+        self.conn.autocommit = False
 
     def close(self):
         self.conn.close()
···
         _ = cursor.execute(f"PRAGMA user_version = {version}")
         self.conn.commit()
 
-    def get_migrations(self) -> list[tuple[int, Path]]:
-        if not self.migrations_folder.exists():
-            return []
-
-        files: list[tuple[int, Path]] = []
-        for f in self.migrations_folder.glob("*.sql"):
-            try:
-                version = int(f.stem.split("_")[0])
-                files.append((version, f))
-            except (ValueError, IndexError):
-                LOGGER.warning("Warning: Skipping invalid migration file: %", f.name)
-
-        return sorted(files, key=lambda x: x[0])
-
-    def apply_migration(self, version: int, path: Path):
-        with open(path, "r") as f:
-            sql = f.read()
-
-        cursor = self.conn.cursor()
+    def apply_migration(self, version: int, filename: str, migration: Callable[[sqlite3.Connection], None]):
         try:
-            _ = cursor.executescript(sql)
+            _ = migration(self.conn)
             self.set_version(version)
-            LOGGER.info("Applied migration: %s", path.name)
+            self.conn.commit()
+            LOGGER.info("Applied migration: %s..", filename)
         except sqlite3.Error as e:
             self.conn.rollback()
-            raise Exception(f"Error applying migration {path.name}: {e}")
+            raise Exception(f"Error applying migration {filename}: {e}")
 
     def migrate(self):
         current_version = self.get_version()
-        migrations = self.get_migrations()
+        from migrations._registry import load_migrations
+        migrations = load_migrations(self.migrations_folder)
 
         if not migrations:
             LOGGER.warning("No migration files found.")
···
             LOGGER.info("No pending migrations.")
             return
 
-        for version, filepath in pending:
-            self.apply_migration(version, filepath)
+        for version, filename, migration in pending:
+            self.apply_migration(version, filename, migration)

migrations/001_initdb.sql

···
-CREATE TABLE IF NOT EXISTS posts (
-    id          INTEGER UNIQUE PRIMARY KEY AUTOINCREMENT,
-    user        TEXT NOT NULL,
-    service     TEXT NOT NULL,
-    identifier  TEXT NOT NULL,
-    parent      INTEGER NULL REFERENCES posts(id),
-    root        INTEGER NULL REFERENCES posts(id),
-    reposted    INTEGER NULL REFERENCES posts(id),
-    extra_data  TEXT NULL
-);
-
-CREATE TABLE IF NOT EXISTS mappings (
-    original INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
-    mapped   INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
-    UNIQUE(original, mapped)
-);

migrations/001_initdb_v1.py

···
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection):
+    _ = conn.execute("""
+        CREATE TABLE IF NOT EXISTS posts (
+            id         INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id    TEXT NOT NULL,
+            service    TEXT NOT NULL,
+            identifier TEXT NOT NULL,
+            parent_id  INTEGER NULL REFERENCES posts(id) ON DELETE SET NULL,
+            root_id    INTEGER NULL REFERENCES posts(id) ON DELETE SET NULL
+        );
+    """)
+    _ = conn.execute("""
+        CREATE TABLE IF NOT EXISTS mappings (
+            original_post_id INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
+            mapped_post_id   INTEGER NOT NULL
+        );
+    """)
+    pass

migrations/002_add_indexes.sql

···
-CREATE INDEX IF NOT EXISTS idx_posts_service_user_identifier
-ON posts (service, user, identifier);
-
-CREATE UNIQUE INDEX IF NOT EXISTS ux_mappings_original_mapped
-ON mappings (original, mapped);

migrations/002_add_reposted_column_v1.py

···
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection):
+    columns = conn.execute("PRAGMA table_info(posts)")
+    column_names = [col[1] for col in columns]
+    if "reposted_id" not in column_names:
+        _ = conn.execute("""
+            ALTER TABLE posts
+            ADD COLUMN reposted_id INTEGER NULL REFERENCES posts(id) ON DELETE SET NULL
+        """)

migrations/003_add_extra_data_column_v1.py

···
+import json
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection):
+    columns = conn.execute("PRAGMA table_info(posts)")
+    column_names = [col[1] for col in columns]
+    if "extra_data" not in column_names:
+        _ = conn.execute("""
+            ALTER TABLE posts
+            ADD COLUMN extra_data TEXT NULL
+        """)
+
+    # migrate old bsky identifiers from json to uri as id and cid in extra_data
+    data = conn.execute("SELECT id, identifier FROM posts WHERE service = 'https://bsky.app';").fetchall()
+    rewrites: list[tuple[str, str, int]] = []
+    for row in data:
+        if row[1][0] == '{' and row[1][-1] == '}':
+            data = json.loads(row[1])
+            rewrites.append((data['uri'], json.dumps({'cid': data['cid']}), row[0]))
+    if rewrites:
+        _ = conn.executemany("UPDATE posts SET identifier = ?, extra_data = ? WHERE id = ?;", rewrites)

migrations/004_initdb_next.py

···
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection):
+    cursor = conn.cursor()
+
+    old_posts = cursor.execute("SELECT * FROM posts;").fetchall()
+    old_mappings = cursor.execute("SELECT * FROM mappings;").fetchall()
+
+    _ = cursor.execute("DROP TABLE posts;")
+    _ = cursor.execute("DROP TABLE mappings;")
+
+    _ = cursor.execute("""
+        CREATE TABLE posts (
+            id          INTEGER UNIQUE PRIMARY KEY AUTOINCREMENT,
+            user        TEXT NOT NULL,
+            service     TEXT NOT NULL,
+            identifier  TEXT NOT NULL,
+            parent      INTEGER NULL REFERENCES posts(id),
+            root        INTEGER NULL REFERENCES posts(id),
+            reposted    INTEGER NULL REFERENCES posts(id),
+            extra_data  TEXT NULL
+        );
+    """)
+
+    _ = cursor.execute("""
+        CREATE TABLE mappings (
+            original INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
+            mapped   INTEGER NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
+            UNIQUE(original, mapped)
+        );
+    """)
+
+    for old_post in old_posts:
+        _ = cursor.execute(
+            """
+            INSERT INTO posts (id, user, service, identifier, parent, root, reposted, extra_data)
+            VALUES (:id, :user_id, :service, :identifier, :parent_id, :root_id, :reposted_id, :extra_data)
+        """,
+            dict(old_post),
+        )
+
+    for mapping in old_mappings:
+        original, mapped = mapping["original_post_id"], mapping["mapped_post_id"]
+        _ = cursor.execute(
+            "INSERT OR IGNORE INTO mappings (original, mapped) VALUES (?, ?)",
+            (original, mapped),
+        )
+        _ = cursor.execute(
+            "INSERT OR IGNORE INTO mappings (original, mapped) VALUES (?, ?)",
+            (mapped, original),
+        )

migrations/005_add_indexes.py

···
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection):
+    _ = conn.execute("""
+        CREATE INDEX IF NOT EXISTS idx_posts_service_user_identifier
+        ON posts (service, user, identifier);
+    """)
+    _ = conn.execute("""
+        CREATE UNIQUE INDEX IF NOT EXISTS ux_mappings_original_mapped
+        ON mappings (original, mapped);
+    """)

migrations/_registry.py

···
+import importlib.util
+from pathlib import Path
+import sqlite3
+from typing import Callable
+
+
+def load_migrations(path: Path) -> list[tuple[int, str, Callable[[sqlite3.Connection], None]]]:
+    migrations: list[tuple[int, str, Callable[[sqlite3.Connection], None]]] = []
+    migration_files = sorted(
+        [f for f in path.glob("*.py") if not f.stem.startswith("_")]
+    )
+
+    for filepath in migration_files:
+        filename = filepath.stem
+        version_str = filename.split("_")[0]
+
+        try:
+            version = int(version_str)
+        except ValueError:
+            raise ValueError('migrations must start with a number!!')
+
+        spec = importlib.util.spec_from_file_location(filepath.stem, filepath)
+        if not spec or not spec.loader:
+            raise Exception(f"Failed to load spec from file: {filepath}")
+
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+
+        if hasattr(module, "migrate"):
+            migrations.append((version, filename, module.migrate))
+        else:
+            raise ValueError(f"Migration {filepath.name} missing 'migrate' function")
+
+    migrations.sort(key=lambda x: x[0])
+    return migrations