Creating transactions inside async fixtures with pytest

[dantownsend]

In an ideal world with pytest-asyncio we could do this:

@pytest.fixture(scope="function", autouse=True)
async def piccolo_transaction(event_loop):
    DB: PostgresEngine = engine_finder()
    async with DB.transaction():
        yield

But pytest-asyncio has an issue where contextvars aren't passed from an async fixture, and Piccolo relies on contextvars to track the current transaction.

https://github.com/pytest-dev/pytest-asyncio/issues/127

[dantownsend]

Work around for now:

@pytest.fixture(scope="function", autouse=True)
async def piccolo_transaction(event_loop):
    """
    It should be as simple as::

        async with DB.transaction():
            yield

    However, pytest-asyncio doesn't pass down contextvars, which is how Piccolo
    handles transactions.

    https://github.com/pytest-dev/pytest-asyncio/issues/127

    For now, this is the work around.

    """
    DB: PostgresEngine = engine_finder()

    connection = await DB.get_new_connection()

    transaction = DB.transaction()
    transaction.connection = connection
    transaction.transaction = transaction.connection.transaction()
    await transaction.begin()

    class TransactionProxy:
        def get(self):
            return transaction

    DB.current_transaction = TransactionProxy()

    yield

    await transaction.rollback()
    await connection.close()

[sylvainOL]

Hello @dantownsend,

do you think this is still relevant as it seems some MR have been merged to fix it (but https://github.com/pytest-dev/pytest-asyncio/issues/127 is still open so I'm not sure :/)

thanks!

[dantownsend]

@sylvainOL This is still the work around I'm using at the moment. I occasionally check in on pytest-asyncio to see if the problem is resolved, but honestly I'm not sure.

It looks like they made some changes in 0.23.0, but I haven't had a chance to check whether this resolves the problem.

[sylvainOL]

thanks for the quick reply ! I'll continue to use it then.

Thnaks again!