From 1e70462c2bd68078e4777798ccc478305dd3504e Mon Sep 17 00:00:00 2001
From: Fusion <noreply@runfusion.ai>
Date: Mon, 18 May 2026 23:09:44 +0200
Subject: [PATCH] fix: use sync engine for alembic migration operations

SQLAlchemy 2.0 async engines don't support the sync context manager
protocol needed by alembic. Create a separate sync engine (using
psycopg2) for migration operations while keeping async engine for
application queries.

- Add psycopg2-binary dependency
- Rename async connection variable to avoid mypy confusion
- Use sync engine for MigrationContext and alembic commands
---
 apps/api/pyproject.toml  |  1 +
 apps/api/src/database.py | 72 +++++++++++++++++++++++-----------------
 2 files changed, 43 insertions(+), 30 deletions(-)

diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml
index 3d49f67..d5b71d9 100644
--- a/apps/api/pyproject.toml
+++ b/apps/api/pyproject.toml
@@ -8,6 +8,7 @@ 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 adbf470..1bc93d2 100644
--- a/apps/api/src/database.py
+++ b/apps/api/src/database.py
@@ -4,6 +4,7 @@ import logging
 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
 
@@ -28,6 +29,15 @@ 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,
@@ -42,44 +52,46 @@ async def init_database(
     """
     for attempt in range(1, max_retries + 1):
         try:
-            # Test basic connectivity
+            # Test basic connectivity (async)
             from sqlalchemy import text
-            conn = await engine.connect()
+            test_conn = await engine.connect()
             try:
-                await conn.execute(text("SELECT 1"))
+                await test_conn.execute(text("SELECT 1"))
             finally:
-                await conn.close()
+                await test_conn.close()
 
             logger.info("Database connection established.")
 
-            # Check if migrations are already complete
-            def _check_migrations():
-                alembic_cfg = Config(alembic_ini_path)
-                script = command.ScriptDirectory.from_config(alembic_cfg)
-                with engine.connect() as conn:
-                    context = MigrationContext.configure(conn)
+            # Use sync engine for alembic operations
+            sync_url = _get_sync_database_url()
+            sync_engine = create_engine(sync_url, poolclass=NullPool)
+
+            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()
-                    return current_rev == head_rev, current_rev, head_rev
-
-            is_current, current_rev, head_rev = await asyncio.get_event_loop().run_in_executor(
-                None, _check_migrations
-            )
-
-            if is_current:
-                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 (sync call in executor)
-            def _run_migrations():
-                alembic_cfg = Config(alembic_ini_path)
-                command.upgrade(alembic_cfg, "head")
-
-            await asyncio.get_event_loop().run_in_executor(None, _run_migrations)
-            logger.info("Database migrations completed successfully.")
-            return True
+                    
+                    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()
 
         except Exception as exc:
             error_msg = str(exc).lower()