"""Alembic migration environment — async-compatible. At runtime the app uses ``postgresql+asyncpg://``. Alembic's CLI is synchronous, so we derive a *sync* psycopg2 URL from the same DATABASE_URL env var by replacing the driver prefix. Run migrations with: alembic upgrade head """ from __future__ import annotations import asyncio import os import re from logging.config import fileConfig from alembic import context from sqlalchemy import pool from sqlalchemy.ext.asyncio import create_async_engine # Alembic Config object (gives access to alembic.ini values). config = context.config # Set up Python logging from alembic.ini. if config.config_file_name is not None: fileConfig(config.config_file_name) # Import the Base so that Alembic can detect model changes for --autogenerate. from app.models import Base # noqa: E402 target_metadata = Base.metadata def _sync_url(async_url: str) -> str: """Convert an asyncpg URL to a psycopg2 URL for Alembic CLI.""" return re.sub(r"postgresql\+asyncpg", "postgresql+psycopg2", async_url) def _get_url() -> str: db_url = os.environ.get("DATABASE_URL", "") if not db_url: # Fall back to settings if env var not set directly. from app.config.settings import settings # noqa: PLC0415 db_url = settings.DATABASE_URL return _sync_url(db_url) def run_migrations_offline() -> None: """Emit SQL without a live DB connection.""" url = _get_url() context.configure( url=url, target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}, compare_type=True, ) with context.begin_transaction(): context.run_migrations() def do_run_migrations(connection): # type: ignore[no-untyped-def] context.configure( connection=connection, target_metadata=target_metadata, compare_type=True, ) with context.begin_transaction(): context.run_migrations() async def run_migrations_online_async() -> None: """Run migrations against a live DB using the async engine.""" async_url = os.environ.get("DATABASE_URL", "") if not async_url: from app.config.settings import settings # noqa: PLC0415 async_url = settings.DATABASE_URL connectable = create_async_engine(async_url, poolclass=pool.NullPool) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) await connectable.dispose() def run_migrations_online() -> None: asyncio.run(run_migrations_online_async()) if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()