Alembic can't run migrations in async mode if loop already exists.

[DLogunoff]

Describe the bug

Alembic docs gives a snippet on how to run migrations in async mode. This snippet only works if event loop do not exist and asyncio.run() will create a new one.

In my current setup I'm trying to run migrations to create a db for (async) tests. So the loop is already exists and I can't use asyncio.run(). So I've tried to use loop.create_task(run_migrations_online()). This worked till context.configure(...) line of code.

Expected behavior

It would be great if alembic migration could be ran in async mode if loop is already exist.

To Reproduce

1. Run a postgres server
2. Run in cmd: alembic init -t async migrations
3. Change asyncio.run(run_migrations_online()) to asyncio.get_event_loop().create_task(run_migrations_online())
4. Create main.py file with a simple async code (I'll paste it here in a section below)
5. Launch main,py

main.py

import asyncio

from alembic import command
from alembic.config import Config


async def main():
    alembic_cfg = Config("./alembic.ini")
    command.upgrade(alembic_cfg, "head")

    await asyncio.sleep(20)


asyncio.run(main())

env.py

config.set_main_option("sqlalchemy.url", "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres")
if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.get_event_loop().create_task(run_migrations_online())

Error

AttributeError: 'NoneType' object has no attribute 'configure'

Versions.

• OS: ubuntu 22.04
• Python: 3.10, 3.11
• Alembic: 1.12
• SQLAlchemy: 2.0
• Database: Postgres
• DBAPI:

Additional context

So there you can find a zip archive with code. There is an extra comments, related to this problem

alembic-issue.zip

In this exact code it's possible to run a migrations via asyncio.run() (could someone explain why this happens? I have same thing in my project, but it happens randomly - one out of 5 runs the migrations will execute, other runs will be ended with asyncio exception about the already running event loop).

Have a nice day!

[zzzeek]

it's not clear how you are invoking Alembic commands, which themselves are not awaitable, however any sync-style function can be run in an asynchronous context using SQLAlchemy's run_sync() services, see https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#running-synchronous-methods-and-functions-under-asyncio

[DLogunoff]

Hi.

We can ignore my main.py file.
The main prerequisite for this problem - alembic config called from somewhere and we are already in async event loop.
The only way I know to run coroutine in existing event loop is loop.create_task.
So in the moment when run_async_migrations() is being executed as a task, the alembic.context loses all his attributes - context.config and context.script. That's why it's impossible to execute migrations from asyncio task

[CaselIT]

Hi,

In my current setup I'm trying to run migrations to create a db for (async) tests. So the loop is already exists and I can't use asyncio.run().

standard asyncio rules apply, you can't have more than one loop running on a thread. Can't you use https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_until_complete instead of asyncio.run if you already have a loop?

[DLogunoff]

Hi.

No, I can't.
So there is the source code of loop.run_until_complete(). It has self._check_running() method call. So there is the source of this method.
It become clear that we can't use loop.run_until_complete() if loop already exists and running.

[CaselIT]

right, as always the asyncio api is great... You can maybe catch the runtime error and call run_async_migrations directly.

all this suggest that you are trying to use alembic in a way that's not meant to be used.
with asyncio you have to really do what they want, or it's an exercise in frustration

[DLogunoff]

all this suggest that you are trying to use alembic in a way that's not meant to be used.

Yeah, that's the reason I guess.

[DLogunoff]

I found a workaround but I don't really like it.

def run_migrations_online_sync():
    # I have to delete +asyncpg from DSN
    conf = config.get_section(config.config_ini_section)
    conf["sqlalchemy.url"] = conf["sqlalchemy.url"].replace("+asyncpg", "")
    connectable = engine_from_config(
        conf,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    # This doesn't work.
    # connectable = async_engine_from_config(
    #     config.get_section(config.config_ini_section),
    #     prefix="sqlalchemy.",
    #     poolclass=pool.NullPool,
    # ).sync_engine

    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online_sync()

In both cases (if I create sync engine with +asycnpg in dsn and if I create async engine and take sync_engine from it) the code is fails with the following exception:

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)

[armanddidierjean]

I was in similar situation as yours and got the same issues. Exploring Alembic documentation I found Programmatic API use (connection sharing) With Asyncio which gave me a solution. This should also correspond to @zzzeek answer.

You could edit env.py:

def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""

    connectable = config.attributes.get("connection", None)

    if connectable is None:
        asyncio.run(run_async_migrations())
    else:
        do_run_migrations(connectable)

and use this main.py

import asyncio
from alembic import command
from alembic.config import Config
from sqlalchemy.ext.asyncio import create_async_engine


def run_upgrade(connection, cfg):
    cfg.attributes["connection"] = connection
    command.upgrade(cfg, "head")

async def main():
    alembic_cfg = Config("./alembic.ini")

    async_engine = create_async_engine(
        "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres", echo=True
    )
    async with async_engine.begin() as conn:
        await conn.run_sync(run_upgrade, alembic_cfg)

    await asyncio.sleep(20)

asyncio.run(main())

Hoping this may help you