ef8285e583
On boot, run_migrations() checks the Alembic revision: - no revision + tables exist → stamp head (db.sql was run manually) - no revision + no tables → upgrade head (fresh install) - revision present → upgrade head (no-op if already current) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import logging
|
|
import os
|
|
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from alembic.runtime.migration import MigrationContext
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
logger = logging.getLogger("pops")
|
|
|
|
|
|
def _db_url() -> str:
|
|
return (
|
|
f"mysql+pymysql://{os.environ['MYSQL_USER']}:{os.environ['MYSQL_PASSWORD']}"
|
|
f"@{os.environ['MYSQL_HOST']}:{os.environ.get('MYSQL_PORT', 3306)}"
|
|
f"/{os.environ['MYSQL_DATABASE']}"
|
|
)
|
|
|
|
|
|
engine = create_engine(_db_url(), pool_pre_ping=True)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
|
|
|
|
def get_db():
|
|
db: Session = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def run_migrations() -> None:
|
|
alembic_cfg = Config("alembic.ini")
|
|
with engine.connect() as conn:
|
|
current_rev = MigrationContext.configure(conn).get_current_revision()
|
|
if current_rev is None:
|
|
if "printers" in inspect(engine).get_table_names():
|
|
logger.info("Existing schema detected, stamping Alembic to head")
|
|
command.stamp(alembic_cfg, "head")
|
|
else:
|
|
logger.info("Fresh database, running migrations")
|
|
command.upgrade(alembic_cfg, "head")
|
|
else:
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
|
|
def check_db() -> tuple[bool, str]:
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1"))
|
|
return True, "ok"
|
|
except Exception as e:
|
|
return False, str(e)
|