feat: [US-001] Add automatic Alembic migrations at backend startup

- Add alembic.ini config for data_entry module
- Add migrations.py helper with:
  - run_migrations(): runs pending migrations at startup (returns True/False)
  - get_current_revision(): returns current schema version for /health
- Integrate migrations in main.py startup_event() BEFORE init_data_entry_db()
- Add data_entry_schema_version to /health endpoint

If migrations fail, backend continues with WARNING (not crash).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-01-28 18:00:01 +00:00
parent 8567f893bc
commit dc1711acd0
3 changed files with 227 additions and 0 deletions

View File

@@ -0,0 +1,94 @@
# Alembic configuration for Data Entry module
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration files
file_template = %%(year)d%%(month).2d%%(day).2d_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
# version path separator
# version_path_separator = :
# set to 'true' to search source files recursively
# in each "version_locations" directory
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# SQLite database URL - will be overridden by env.py using SQLITE_DATABASE_PATH env var
sqlalchemy.url = sqlite:///data/receipts/receipts.db
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - disabled
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -q
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

View File

@@ -0,0 +1,131 @@
"""
Alembic migrations helper for Data Entry module.
Provides automatic migration execution at backend startup.
"""
import logging
import os
from pathlib import Path
logger = logging.getLogger(__name__)
def run_migrations() -> bool:
"""
Run pending Alembic migrations at startup.
Returns:
True if migrations ran successfully (or no pending migrations),
False if migrations failed (backend should continue with WARNING).
"""
try:
from alembic.config import Config
from alembic import command
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine
# Get the path to alembic.ini
data_entry_module = Path(__file__).parent.parent
alembic_ini_path = data_entry_module / "alembic.ini"
if not alembic_ini_path.exists():
logger.warning(f"[MIGRATIONS] alembic.ini not found at {alembic_ini_path}")
return False
# Get database path from environment or default
db_path = Path(os.getenv(
"SQLITE_DATABASE_PATH",
"data/receipts/receipts.db"
)).resolve()
# Ensure database directory exists
db_path.parent.mkdir(parents=True, exist_ok=True)
# Create Alembic config
alembic_cfg = Config(str(alembic_ini_path))
# Override database URL
sync_db_url = f"sqlite:///{db_path}"
alembic_cfg.set_main_option("sqlalchemy.url", sync_db_url)
# Set script location relative to alembic.ini
alembic_cfg.set_main_option(
"script_location",
str(data_entry_module / "migrations")
)
# Get current revision before upgrade
engine = create_engine(sync_db_url)
with engine.connect() as connection:
context = MigrationContext.configure(connection)
current_rev = context.get_current_revision()
engine.dispose()
logger.info(f"[MIGRATIONS] Current revision: {current_rev or 'None (fresh database)'}")
logger.info(f"[MIGRATIONS] Database path: {db_path}")
# Run upgrade to head
logger.info("[MIGRATIONS] Checking for pending migrations...")
command.upgrade(alembic_cfg, "head")
# Get new revision after upgrade
engine = create_engine(sync_db_url)
with engine.connect() as connection:
context = MigrationContext.configure(connection)
new_rev = context.get_current_revision()
engine.dispose()
if current_rev != new_rev:
logger.info(f"[MIGRATIONS] Applied: {current_rev or 'None'} -> {new_rev}")
else:
logger.info(f"[MIGRATIONS] No pending migrations. Current: {new_rev}")
return True
except ImportError as e:
logger.warning(f"[MIGRATIONS] Alembic not installed: {e}")
logger.warning("[MIGRATIONS] Skipping migrations - install alembic to enable")
return False
except Exception as e:
logger.error(f"[MIGRATIONS] Migration error: {e}", exc_info=True)
logger.warning("[MIGRATIONS] Backend will continue without migrations")
return False
def get_current_revision() -> str:
"""
Get the current Alembic revision.
Returns:
Current revision string, or 'unknown' if cannot be determined.
"""
try:
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine
# Get database path from environment or default
db_path = Path(os.getenv(
"SQLITE_DATABASE_PATH",
"data/receipts/receipts.db"
)).resolve()
if not db_path.exists():
return "no_database"
sync_db_url = f"sqlite:///{db_path}"
engine = create_engine(sync_db_url)
with engine.connect() as connection:
context = MigrationContext.configure(connection)
revision = context.get_current_revision()
engine.dispose()
return revision or "none"
except ImportError:
return "alembic_not_installed"
except Exception as e:
logger.debug(f"[MIGRATIONS] Could not get revision: {e}")
return "unknown"