82 lines
No EOL
2.4 KiB
Python
82 lines
No EOL
2.4 KiB
Python
"""Tiny migration runner: applies migrations/NNN_*.sql in order.
|
|
|
|
Tracks applied migrations in a `schema_migrations` table.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import psycopg
|
|
|
|
from app.config import DATABASE_URL, MIGRATIONS_DIR
|
|
|
|
|
|
def _ensure_migrations_table(conn: psycopg.Connection) -> None:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
id int PRIMARY KEY,
|
|
filename text NOT NULL,
|
|
applied_at timestamptz NOT NULL DEFAULT now()
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def list_migration_files() -> list[tuple[int, Path]]:
|
|
"""Return sorted (number, path) for all NNN_*.sql files."""
|
|
pattern = re.compile(r"^(\d{3})_.*\.sql$")
|
|
result: list[tuple[int, Path]] = []
|
|
if not MIGRATIONS_DIR.exists():
|
|
return result
|
|
for p in sorted(MIGRATIONS_DIR.iterdir()):
|
|
m = pattern.match(p.name)
|
|
if m:
|
|
result.append((int(m.group(1)), p))
|
|
result.sort(key=lambda t: t[0])
|
|
return result
|
|
|
|
|
|
def get_applied(conn: psycopg.Connection) -> set[int]:
|
|
_ensure_migrations_table(conn)
|
|
rows = conn.execute("SELECT id FROM schema_migrations").fetchall()
|
|
return {r[0] for r in rows}
|
|
|
|
|
|
def run_migrations(database_url: str | None = None) -> list[int]:
|
|
"""Apply all pending migrations. Returns list of applied migration numbers."""
|
|
url = database_url or DATABASE_URL
|
|
applied_ids: list[int] = []
|
|
with psycopg.connect(url) as conn:
|
|
applied = get_applied(conn)
|
|
for num, path in list_migration_files():
|
|
if num in applied:
|
|
continue
|
|
sql = path.read_text()
|
|
conn.execute(sql)
|
|
conn.execute(
|
|
"INSERT INTO schema_migrations (id, filename) VALUES (%s, %s)",
|
|
(num, path.name),
|
|
)
|
|
conn.commit()
|
|
applied_ids.append(num)
|
|
return applied_ids
|
|
|
|
|
|
def reset_database(database_url: str | None = None) -> None:
|
|
"""Drop all tables (for tests only) and re-run migrations."""
|
|
url = database_url or DATABASE_URL
|
|
with psycopg.connect(url) as conn:
|
|
conn.execute(
|
|
"""
|
|
DROP TABLE IF EXISTS task_run, outbox, approval, artifact,
|
|
email_suggestion, notification_log,
|
|
application, job_posting, cv_section, profile, schema_migrations
|
|
CASCADE
|
|
"""
|
|
)
|
|
conn.commit()
|
|
run_migrations(url) |