Running migrations in pytest

[ZeN220]

Im tried run alembic migrations like this code:

from alembic.command import downgrade, upgrade
from alembic.config import Config as AlembicConfig
from pytest_asyncio import fixture
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import Config


@fixture(scope="function")
async def engine(config: Config) -> AsyncEngine:
    engine = create_async_engine(config.postgres.dsn)
    yield engine
    await engine.dispose()


@fixture(scope="function")
async def session_maker(
    engine: AsyncEngine,
) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(engine, expire_on_commit=False)


@fixture(scope="function")
async def session(
    session_maker: async_sessionmaker,
    engine: AsyncEngine,
    alembic_config: AlembicConfig,
) -> AsyncSession:
    upgrade(alembic_config, "head")
    async with session_maker() as session:
        yield session
        await session.rollback()
    downgrade(alembic_config, "base")

but because under hood alembic.commands.upgrade/downgrade uses asyncio.run, i catch an error:

RuntimeError: asyncio.run() cannot be called from a running event loop

how i can effective runalembic migrations in my tests?

[CaselIT]

Hi,

Asycio makes everything complex...
You would need to update your env.py to change the asyncio.run to check if there is an existing even loop to re-use that

[ZeN220]

i still havent come up with adequate solution to run this without using the loop. do u have any ideas?

[CaselIT]

something like this?

try:
  loop = asyncio.get_running_loop()
except RuntimeError:
  loop = None
if loop:
  loop.run_until_complete(run_async_migrations()) 
else:
  asyncio.run(run_async_migrations())

[ZeN220]

yea, im tried like this. but loop.run_until_complete doesnt work in running loop

[CaselIT]

right python really don't want to make you use asyncio from sync code...

You can probably use the same functionality used by sqlalchemy.
Try something like this (not 100% sure it will work)
In the test:

    async with engine.connect() as c:
        await c.run_sync(lambda _: upgrade(alembic_config, "head"))
    async with session_maker() as session:
        yield session
        await session.rollback()
    async with engine.connect() as c:
        await c.run_sync(lambda _: downgrade(alembic_config, "base"))

change the run_migrations_online to do the following

from sqlalchemy.util.concurrency import await_only, in_greenlet

if in_greenlet():
    await_only(run_async_migrations())
else:
    asyncio.run(run_async_migrations())

@zzzeek we should probably export in_greenlet from sqlalchemy.util