Auto-stamp or migrate on startup

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>
This commit is contained in:
Martin Hohenberg
2026-06-18 21:21:51 +02:00
parent d092c19b41
commit ef8285e583
2 changed files with 24 additions and 2 deletions

View File

@@ -1,8 +1,14 @@
import logging
import os
from sqlalchemy import create_engine, text
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 (
@@ -24,6 +30,21 @@ def get_db():
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:

View File

@@ -3,7 +3,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db import check_db
from app.db import check_db, run_migrations
from app.routers import filaments, pricing, print_jobs, printers, todos
logger = logging.getLogger("pops")
@@ -14,6 +14,7 @@ async def lifespan(app: FastAPI):
ok, msg = check_db()
if ok:
logger.info("Database connection OK")
run_migrations()
else:
logger.warning("Database unreachable: %s", msg)
yield