87 lines
2.2 KiB
Python
Executable File
87 lines
2.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Apply SQL migrations stored in ./migrations.
|
|
|
|
Usage:
|
|
python scripts/apply_migrations.py
|
|
python scripts/apply_migrations.py --dry-run
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
import os
|
|
import sys
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
os.chdir(PROJECT_ROOT)
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MIGRATIONS_DIR = ROOT / "migrations"
|
|
|
|
|
|
def ensure_ledger(conn) -> None:
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
)
|
|
"""))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--dry-run", action="store_true", help="List pending migrations without applying them.")
|
|
args = parser.parse_args()
|
|
|
|
files = sorted(MIGRATIONS_DIR.glob("*.sql"))
|
|
if not files:
|
|
print("No migrations found.")
|
|
return 0
|
|
|
|
with engine.begin() as conn:
|
|
ensure_ledger(conn)
|
|
applied = {
|
|
row[0]
|
|
for row in conn.execute(text("SELECT version FROM schema_migrations"))
|
|
}
|
|
|
|
pending = []
|
|
for path in files:
|
|
version = path.stem.split("_", 1)[0]
|
|
if version not in applied:
|
|
pending.append((version, path))
|
|
|
|
if not pending:
|
|
print("No pending migrations.")
|
|
return 0
|
|
|
|
print("Pending migrations:")
|
|
for version, path in pending:
|
|
print(f"- {version}: {path.name}")
|
|
|
|
if args.dry_run:
|
|
return 0
|
|
|
|
for version, path in pending:
|
|
sql = path.read_text()
|
|
print(f"Applying {path.name}...")
|
|
conn.execute(text(sql))
|
|
conn.execute(text("""
|
|
INSERT INTO schema_migrations(version, name)
|
|
VALUES (:version, :name)
|
|
ON CONFLICT (version) DO NOTHING
|
|
"""), {"version": version, "name": path.name})
|
|
|
|
print("Migrations applied.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|