28 lines
971 B
Python
28 lines
971 B
Python
"""Fail deployment if migrations or runtime table definitions are missing."""
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
from sqlalchemy import select, text
|
|
from src.core.database import engine, Base
|
|
from src.core import models
|
|
|
|
|
|
async def main():
|
|
expected = set(ScriptDirectory.from_config(Config("alembic.ini")).get_heads())
|
|
async with engine.connect() as connection:
|
|
actual = set((await connection.scalars(text("SELECT version_num FROM alembic_version"))).all())
|
|
if actual != expected:
|
|
raise RuntimeError("Database migrations are not at the release head")
|
|
for table in Base.metadata.sorted_tables:
|
|
await connection.execute(select(table).limit(0))
|
|
await engine.dispose()
|
|
print("Database schema verified")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|