115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Mapping
|
|
|
|
from sqlalchemy import inspect, text
|
|
|
|
from services.shared.db import engine
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
MIGRATIONS_DIR = ROOT / "migrations" / "sql"
|
|
_MODE_ENV = "SCHEMA_MANAGEMENT_MODE"
|
|
_VALID_MODES = {"legacy", "migrations"}
|
|
|
|
|
|
def database_backend_name(db_engine=engine) -> str:
|
|
return db_engine.url.get_backend_name()
|
|
|
|
|
|
def migration_suffix(db_engine=engine) -> str:
|
|
return "postgres" if database_backend_name(db_engine) == "postgresql" else "sqlite"
|
|
|
|
|
|
def list_migration_files(*, db_engine=engine, migrations_dir: Path = MIGRATIONS_DIR) -> list[Path]:
|
|
return sorted(migrations_dir.glob(f"*_{migration_suffix(db_engine)}.sql"))
|
|
|
|
|
|
def resolve_schema_management_mode(
|
|
raw_mode: str | None,
|
|
*,
|
|
backend_name: str,
|
|
) -> str:
|
|
normalized = str(raw_mode or "").strip().lower()
|
|
if not normalized:
|
|
normalized = "legacy" if backend_name == "sqlite" else "migrations"
|
|
if normalized not in _VALID_MODES:
|
|
allowed = ", ".join(sorted(_VALID_MODES))
|
|
raise RuntimeError(f"{_MODE_ENV} must be one of: {allowed}")
|
|
if backend_name != "sqlite" and normalized != "migrations":
|
|
raise RuntimeError(
|
|
f"{_MODE_ENV}=legacy is supported only for SQLite. "
|
|
"Run `python scripts/migrate_core_db.py` and use SCHEMA_MANAGEMENT_MODE=migrations."
|
|
)
|
|
return normalized
|
|
|
|
|
|
def schema_management_mode(
|
|
*,
|
|
db_engine=engine,
|
|
environ: Mapping[str, str] | None = None,
|
|
) -> str:
|
|
source = environ or os.environ
|
|
return resolve_schema_management_mode(
|
|
source.get(_MODE_ENV),
|
|
backend_name=database_backend_name(db_engine),
|
|
)
|
|
|
|
|
|
def ensure_schema_migrations_table(*, db_engine=engine) -> None:
|
|
with db_engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version TEXT PRIMARY KEY,
|
|
applied_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
|
|
|
|
def has_schema_migrations_table(*, db_engine=engine) -> bool:
|
|
return inspect(db_engine).has_table("schema_migrations")
|
|
|
|
|
|
def applied_migration_versions(*, db_engine=engine) -> set[str]:
|
|
if not has_schema_migrations_table(db_engine=db_engine):
|
|
return set()
|
|
with db_engine.begin() as conn:
|
|
rows = conn.execute(text("SELECT version FROM schema_migrations")).fetchall()
|
|
return {str(row[0]) for row in rows}
|
|
|
|
|
|
def missing_migration_versions(*, db_engine=engine, migrations_dir: Path = MIGRATIONS_DIR) -> list[str]:
|
|
expected = [path.name for path in list_migration_files(db_engine=db_engine, migrations_dir=migrations_dir)]
|
|
applied = applied_migration_versions(db_engine=db_engine)
|
|
return [name for name in expected if name not in applied]
|
|
|
|
|
|
def validate_schema_migrations_applied(*, db_engine=engine, migrations_dir: Path = MIGRATIONS_DIR) -> None:
|
|
expected = list_migration_files(db_engine=db_engine, migrations_dir=migrations_dir)
|
|
if not expected:
|
|
raise RuntimeError(f"No migration files found for backend {database_backend_name(db_engine)} in {migrations_dir}")
|
|
|
|
if not has_schema_migrations_table(db_engine=db_engine):
|
|
raise RuntimeError(
|
|
"Database schema is not initialized via migrations. "
|
|
"Run `python scripts/migrate_core_db.py` before starting services."
|
|
)
|
|
|
|
missing = missing_migration_versions(db_engine=db_engine, migrations_dir=migrations_dir)
|
|
if not missing:
|
|
return
|
|
|
|
preview = ", ".join(missing[:5])
|
|
remainder = "" if len(missing) <= 5 else f" ... (+{len(missing) - 5} more)"
|
|
raise RuntimeError(
|
|
"Database schema is behind the checked-in migrations. "
|
|
f"Missing: {preview}{remainder}. "
|
|
"Run `python scripts/migrate_core_db.py` before starting services."
|
|
)
|