Move app settings schema into SQL migration

repub/model.py

@@ -19,6 +19,7 @@ from peewee import (
     SqliteDatabase,
     TextField,
 )
+from playhouse.migrate import SchemaMigrator, migrate
 
 DEFAULT_DB_PATH = Path("republisher.db")
 DATABASE_PRAGMAS = {
@@ -79,45 +80,50 @@ def initialize_database(db_path: str | Path | None = None) -> Path:
     database.init(str(resolved_path), pragmas=DATABASE_PRAGMAS)
     database.connect(reuse_if_open=True)
     try:
-        connection = database.connection()
         for path in schema_paths():
-            connection.executescript(path.read_text(encoding="utf-8"))
-        _ensure_schema(connection)
+            database.connection().executescript(path.read_text(encoding="utf-8"))
+        _run_legacy_migrations()
     finally:
         database.close()
 
     return resolved_path
 
 
-def _ensure_schema(connection: Any) -> None:
-    connection.execute(
-        """
-        CREATE TABLE IF NOT EXISTS app_setting (
-            key TEXT PRIMARY KEY,
-            value TEXT NOT NULL
-        )
-        """
-    )
-
-    job_columns = {
-        row[1] for row in connection.execute("PRAGMA table_info('job')").fetchall()
-    }
+def _run_legacy_migrations() -> None:
+    job_columns = {column.name for column in database.get_columns("job")}
+    operations = []
+    migrator = SchemaMigrator.from_database(database)
     if "convert_images" not in job_columns:
-        connection.execute(
-            """
-            ALTER TABLE job
-            ADD COLUMN convert_images INTEGER NOT NULL DEFAULT 1
-            CHECK (convert_images IN (0, 1))
-            """
+        operations.extend(
+            (
+                migrator.add_column(
+                    "job",
+                    "convert_images",
+                    BooleanField(
+                        default=True,
+                        constraints=[Check("convert_images IN (0, 1)")],
+                    ),
+                ),
+                migrator.add_column_default("job", "convert_images", 1),
+            )
         )
     if "convert_video" not in job_columns:
-        connection.execute(
-            """
-            ALTER TABLE job
-            ADD COLUMN convert_video INTEGER NOT NULL DEFAULT 1
-            CHECK (convert_video IN (0, 1))
-            """
+        operations.extend(
+            (
+                migrator.add_column(
+                    "job",
+                    "convert_video",
+                    BooleanField(
+                        default=True,
+                        constraints=[Check("convert_video IN (0, 1)")],
+                    ),
+                ),
+                migrator.add_column_default("job", "convert_video", 1),
+            )
         )
+    if operations:
+        with database.atomic():
+            migrate(*operations)
 
 
 def source_slug_exists(slug: str) -> bool:

repub/sql/002_app_setting.sql

@@ -0,0 +1,4 @@
+CREATE TABLE IF NOT EXISTS app_setting (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);