fix: use subprocess for alembic migrations to avoid async/sync issues

SQLAlchemy 2.0 async engines conflict with alembic's sync context manager.
Instead of trying to bridge async/sync, use subprocess to run
'alembic upgrade head' directly. This is simpler and more reliable.

- Remove psycopg2-binary dependency (no longer needed)
- Simplify init_database to use subprocess.run()
- Remove all sync engine code
This commit is contained in:
Fusion
2026-05-18 23:16:01 +02:00
parent 1e70462c2b
commit caf73e39ba
2 changed files with 27 additions and 47 deletions
+27 -46
View File
@@ -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()