docekr

app/database_migrations.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from sqlalchemy import inspect, text
+
+from .extensions import db
+
+
+@dataclass(frozen=True)
+class Migration:
+    version: str
+    upgrade: Callable[[], None]
+
+
+MIGRATIONS: tuple[Migration, ...] = ()
+
+def ensure_migration_table() -> None:
+    inspector = inspect(db.engine)
+    if inspector.has_table("schema_migration"):
+        return
+
+    db.session.execute(text("""
+        CREATE TABLE schema_migration (
+            version VARCHAR(100) PRIMARY KEY,
+            applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+        )
+    """))
+    db.session.commit()
+
+
+def migrate_database() -> list[str]:
+    """Uruchamia wyłącznie niewykonane migracje."""
+    ensure_migration_table()
+
+    applied = {
+        row[0]
+        for row in db.session.execute(text("SELECT version FROM schema_migration"))
+    }
+    completed: list[str] = []
+
+    for migration in MIGRATIONS:
+        if migration.version in applied:
+            continue
+
+        try:
+            migration.upgrade()
+            db.session.execute(
+                text("INSERT INTO schema_migration (version) VALUES (:version)"),
+                {"version": migration.version},
+            )
+            db.session.commit()
+            completed.append(migration.version)
+        except Exception:
+            db.session.rollback()
+            raise
+
+    return completed
+
+
+def main() -> None:
+    from . import create_app
+
+    app = create_app()
+    with app.app_context():
+        applied = migrate_database()
+
+    if applied:
+        print("Wykonane migracje: " + ", ".join(applied))
+    else:
+        print("Brak nowych migracji.")
+
+
+if __name__ == "__main__":
+    main()