75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Run database migrations using Alembic.
|
|
|
|
This script should be run before starting the application to ensure
|
|
the database schema is up to date. It can be run standalone or called
|
|
from run_api.py before starting uvicorn.
|
|
|
|
Usage:
|
|
uv run python -m alpine_bits_python.run_migrations
|
|
or
|
|
from alpine_bits_python.run_migrations import run_migrations
|
|
run_migrations()
|
|
"""
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from .logging_config import get_logger
|
|
|
|
_LOGGER = get_logger(__name__)
|
|
|
|
|
|
def run_migrations() -> None:
|
|
"""Run Alembic migrations to upgrade database to latest schema.
|
|
|
|
This function runs 'alembic upgrade head' to apply all pending migrations.
|
|
It will exit the process if migrations fail.
|
|
|
|
Raises:
|
|
SystemExit: If migrations fail
|
|
"""
|
|
_LOGGER.info("Running database migrations...")
|
|
|
|
# Get the project root directory (where alembic.ini is located)
|
|
# Assuming this file is in src/alpine_bits_python/
|
|
project_root = Path(__file__).parent.parent.parent
|
|
|
|
try:
|
|
# Run alembic upgrade head
|
|
result = subprocess.run(
|
|
["alembic", "upgrade", "head"],
|
|
cwd=project_root,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
_LOGGER.info("Database migrations completed successfully")
|
|
_LOGGER.debug("Migration output: %s", result.stdout)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
_LOGGER.error("Failed to run database migrations:")
|
|
_LOGGER.error("Exit code: %d", e.returncode)
|
|
_LOGGER.error("stdout: %s", e.stdout)
|
|
_LOGGER.error("stderr: %s", e.stderr)
|
|
sys.exit(1)
|
|
except FileNotFoundError:
|
|
_LOGGER.error(
|
|
"Alembic not found. Please ensure it's installed: uv pip install alembic"
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Configure basic logging if run directly
|
|
import logging
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
run_migrations()
|
|
print("Migrations completed successfully!")
|