Mateusz Gruszczyński
86 lines
2.2 KiB
Python
86 lines
2.2 KiB
Python
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]
|
|
|
|
|
|
def add_station_access_mode() -> None:
|
|
inspector = inspect(db.engine)
|
|
columns = {column["name"] for column in inspector.get_columns("company_settings")}
|
|
if "station_access_mode" not in columns:
|
|
db.session.execute(text(
|
|
"ALTER TABLE company_settings ADD COLUMN station_access_mode VARCHAR(40) "
|
|
"NOT NULL DEFAULT 'all_prefer_favorites'"
|
|
))
|
|
|
|
|
|
MIGRATIONS: tuple[Migration, ...] = (
|
|
Migration("20260713_company_station_access_mode", add_station_access_mode),
|
|
)
|
|
|
|
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()
|