diff --git a/.claude/settings.json b/.claude/settings.json index 949d40d..e959a79 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -23,6 +23,8 @@ "Bash(ruff format*)", "Bash(mypy*)", "Bash(pip install*)", + "Bash(cd backend && alembic*)", + "Bash(python -m py_compile*)", "Bash(git checkout*)", "Bash(git status)", "Bash(git diff*)", diff --git a/backend/modules/data_entry/alembic.ini b/backend/modules/data_entry/alembic.ini new file mode 100644 index 0000000..800e3b6 --- /dev/null +++ b/backend/modules/data_entry/alembic.ini @@ -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 diff --git a/backend/modules/data_entry/db/migrations.py b/backend/modules/data_entry/db/migrations.py new file mode 100644 index 0000000..28e8a62 --- /dev/null +++ b/backend/modules/data_entry/db/migrations.py @@ -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"