Incompatible with anyio due to event loop keep-alive for session fixtures

[gaborbernat]

anyio creates an event loop during the fixture, and that loop will be reused/kept alive until the fixture is cleaned up (see https://anyio.readthedocs.io/en/stable/testing.html#technical-details). This means that if you have a session scooped fixture (which is quite common, as you might have a DB setup fixture):

@pytest.fixture(scope="session", name="db")
async def _setup_db() -> None:
    if await database_exists(ENGINE.url):  # pragma: no branch
        await drop_database(ENGINE.url)  # pragma: no cover
    await setup_db()

Then this fixture will keep alive the event loop until the end of the session (such fixture requires all your tests using the DB to reuse the same event loop).

pytest-alembic, however, assumes that when its test is running, there is no other event loop running, as it is calling https://github.com/schireson/pytest-alembic/blob/5ff6ca08f553925095389da6048895d41de2b505/src/pytest_alembic/executor.py#L182 which is only allowed to be called when there is no other live event loop.

Instead, pytest-alembic IMHO should check if there is a live event loop and reuse it if so. A working workaround for now would be to create a new thread (that by default has no running loop on it):

    with ThreadPoolExecutor() as thread_pool:
        thread_pool.submit(upgrade, alembic_runner).result()

[DanCardin]

I almost wonder if the thread pool option would be preferable to trying to attach to the running loop. dealing with pytest-asyncio has already been a nightmare, and i definitely feel like I've tried to make it work more integratedly before, and it ran into many (potentially pytest-asyncio specific) issues.

[gaborbernat]

I almost wonder if the thread pool option would be preferable to trying to attach to the running loop. dealing with pytest-asyncio has already been a nightmare, and i definitely feel like I've tried to make it work more integratedly before, and it ran into many (potentially pytest-asyncio specific) issues.

yeah, that could be a good solution

[DanielHabenicht]

@gaborbernat Do you know where to (re)place code if you want to run pytest --test-alembic?

 with ThreadPoolExecutor() as thread_pool:
        thread_pool.submit(upgrade, alembic_runner).result()

[gaborbernat]

I never use that flag, so can't tell.

[DanCardin]

I am struggling to come up with a setup that works enough to produce the failing behavior. Ideally one of you could make a branch that copies or modifies, say, examples/test_async_sqlalchemy (pytest tests/test_runner.py::test_async_sqlalchemy) to the point where it fails in the right way? Or else a much more comprehensive set of code blocks here that enumerate a minimal conftest.py/fixtures setup and a minimal env.py.

With my naive attempt at a new test for asyncio (after uninstalling pytest-asyncio, because it appears to not play wel when they're installed together), it either fully succeeds or fails inside of env.py setup, not the actual tests.

I'll have to look into what precisely anyio is doing, but it seems like the test definitions need to be async functions in order for anyio to do anything with them; and to me that means the default --test-alembic is never going to work

which, i assume, means

async def test_upgrade(alembic):
    tests.test_upgrade(alembic)

only way that will work. I suppose perhaps a --async-test-alembic might be the solution there.

[nat45928]

I found an example of this behavior, I think.

I'm trying to implement some "custom tests" for specific migrations, while using the async test runner like so:

async def test_migration_manual(alembic_runner, alembic_engine: AsyncEngine) -> None:
    # Migrate up to, but not including this new migration
    alembic_runner.migrate_up_before(MIGRATION_ID)

    # preform some database setup prior to
    ...

    # now migrate up to the given revision
    alembic_runner.migrate_up_one()

    # manually create database entries, typically useful for when the current database
    name = "foo"
    ...

    # connect back to the DB to run queries
    async with alembic_engine.connect() as conn:
        result = await conn.execute(text(f"SELECT * FROM foo WHERE id = '0'"))
        db_item = result.one()

        assert db_item.name == name

This runs into the E RuntimeError: asyncio.run() cannot be called from a running event loop from:

../../venv/lib/python3.11/site-packages/pytest_alembic/runner.py:202: in migrate_up_one
    new_revision = self.managed_upgrade(next_revision, current=current)
../../venv/lib/python3.11/site-packages/pytest_alembic/runner.py:150: in managed_upgrade
    self.insert_into(data=before_upgrade_data, revision=current_revision, table=None)
../../venv/lib/python3.11/site-packages/pytest_alembic/runner.py:252: in insert_into
    self.connection_executor.table_insert(
../../venv/lib/python3.11/site-packages/pytest_alembic/executor.py:156: in table_insert
    self.run_task(table_insert, data=data, tablename=tablename, schema=schema)
../../venv/lib/python3.11/site-packages/pytest_alembic/executor.py:183: in run_task
    return asyncio.run(run(self.connection))

Removing async from the test definition works, but then I'm not sure how to query the DB with the async engine without doing a nested async function and a call to asyncio.run from the test body to perform the data checks.

[DanCardin]

I think I might be able to reliably get something failing now, so that's something. I dont seem to be able to successfully bypass the issue (at least within pytest-alembic's own test harness) with a threadpool as the above comment suggests.

I'm anticipating that the solution is going to have to be create_async_alembic_fixture(), which "just" takes the async engine, retrieves a sync interface from it, then runs the rest of the plugin/alembic normally (the main "drawback" being it'd require the env.py to interchangeably work with sync/async like the docs suggest doing already). I might be missing something, but the way anyio seems to instrument the event loop, it's making it very challenging to avoid nested event loop calls into sqlalchemy.

And if that's all it's doing, it begs the question, why can't the migrations/tests just be run synchronously anyway? Afaict, you can create a sync engine with an async driver. So it's just not obvious to me what the value is in having alembic be dealing with asyncio anyways, given that the actual individual alembic per-migration interface is still synchronous too...

(A genuine question, we use async, but we also just have a synchronous driver installed in the migrations context anyways. but presumably we could just use an async driver and continue to have an identical sync setup to what we have today?)

[Lawouach]

(A genuine question, we use async, but we also just have a synchronous driver installed in the migrations context anyways. but presumably we could just use an async driver and continue to have an identical sync setup to what we have today?)

After too much despair, that's what I ended up doing. I run the alembic fixture with a sync driver (psycopg2) and, once the migration is completed, I let the application run with the async driver. That seems to be a much simpler path forward.

[DanCardin]

my impression is you can still use async drivers to create sync engines, so i would expect you to be able to uninstall psycopg2 and use whatever async driver you're using (asyncpg/psycopg?) in the same context, fwiw.