diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml index d5b71d9..3d49f67 100644 --- a/apps/api/pyproject.toml +++ b/apps/api/pyproject.toml @@ -8,7 +8,6 @@ dependencies = [ "uvicorn[standard]>=0.24.0", "sqlalchemy>=2.0.0", "asyncpg>=0.29.0", - "psycopg2-binary>=2.9.0", "alembic>=1.12.0", "pydantic>=2.5.0", "pydantic-settings>=2.1.0", diff --git a/apps/api/src/database.py b/apps/api/src/database.py index 1bc93d2..7f9b5e3 100644 --- a/apps/api/src/database.py +++ b/apps/api/src/database.py @@ -1,10 +1,7 @@ import asyncio import logging +import subprocess -from alembic import command -from alembic.config import Config -from alembic.runtime.migration import MigrationContext -from sqlalchemy import create_engine from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool @@ -29,30 +26,20 @@ engine = create_async_engine( SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) -def _get_sync_database_url() -> str: - """Convert async database URL to sync URL for alembic.""" - sync_url = database_url - # Replace asyncpg with psycopg2 for sync operations - if "+asyncpg" in sync_url: - sync_url = sync_url.replace("+asyncpg", "+psycopg2") - return sync_url - - async def init_database( max_retries: int = 5, retry_delay: float = 2.0, - alembic_ini_path: str = "alembic.ini", ) -> bool: """Initialize the database by running pending migrations. - Retries the connection with exponential backoff to handle - database startup latency in containerized environments. + Uses subprocess to run 'alembic upgrade head' to avoid + async/sync context manager issues with SQLAlchemy 2.0. Returns True if migrations succeeded, False otherwise. """ for attempt in range(1, max_retries + 1): try: - # Test basic connectivity (async) + # Test basic connectivity from sqlalchemy import text test_conn = await engine.connect() try: @@ -62,36 +49,30 @@ async def init_database( logger.info("Database connection established.") - # Use sync engine for alembic operations - sync_url = _get_sync_database_url() - sync_engine = create_engine(sync_url, poolclass=NullPool) + # Run migrations via subprocess + logger.info("Running database migrations...") + result = await asyncio.get_event_loop().run_in_executor( + None, + lambda: subprocess.run( + ["alembic", "upgrade", "head"], + capture_output=True, + text=True, + cwd="/app", + ), + ) - try: - # Check if migrations are already complete - with sync_engine.connect() as sync_conn: - context = MigrationContext.configure(sync_conn) - current_rev = context.get_current_revision() - - alembic_cfg = Config(alembic_ini_path) - script = command.ScriptDirectory.from_config(alembic_cfg) - head_rev = script.get_current_head() - - if current_rev == head_rev: - logger.info("Database is already at the latest migration (%s).", head_rev) - return True - - logger.info( - "Current revision: %s, Head revision: %s. Running migrations...", - current_rev, - head_rev, - ) - - # Run alembic migrations - command.upgrade(alembic_cfg, "head") - logger.info("Database migrations completed successfully.") - return True - finally: - sync_engine.dispose() + if result.returncode == 0: + logger.info("Database migrations completed successfully.") + logger.debug("Alembic output: %s", result.stdout) + return True + else: + logger.error("Migration failed: %s", result.stderr) + if attempt < max_retries: + wait = retry_delay * (2 ** (attempt - 1)) + logger.info("Retrying in %.1f seconds...", wait) + await asyncio.sleep(wait) + else: + return False except Exception as exc: error_msg = str(exc).lower()